/*-------------------------------------------------------------------------
 *
 * reorderbuffer.c
 *	  PostgreSQL 逻辑重放/重排序缓冲区管理
 *
 *
 * Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/reorderbuffer.c
 *
 * NOTES
 *	  此模块接收以写入 WAL 顺序的单个事务部分，并负责将它们重新
 *	  组装成顶级事务大小的部分。当一个事务完全重组 - 通过读取
 *	  事务提交记录来发出信号 - 它将调用输出插件（参见 ReorderBufferCommit()）
 *	  以及单个更改。输出插件依赖于由 snapbuild.c 构建的快照，将其
 *	  传递给我们。
 *
 *	  PostgreSQL 中的事务和子事务/保存点之间立即没有相互链接。
 *	  只有在提交/中止（或特殊的 xact_assignment 记录）时，它们才会
 *	  相互链接。这意味着我们必须将一个顶级事务与其子事务拼接在
 *	  一起。为了高效做到这一点，我们构建一个由各个子事务的 changestreams
 *	  的最小当前 LSN 索引的二进制堆。因为各个流本质上是按 LSN 排
 *	  序的 - 因为我们是从那里构建它们的 - 所以可以通过始终使用堆中
 *	  当前最小 LSN 的子事务轻松地重新组合事务。
 *
 *	  为了应对大事务 - 这些事务可能是可用内存的几倍 - 此模块支持将
 *	  大事务的内容缓冲至磁盘。当重放事务时，单个（子）事务的内容
 *	  将以块的形式从磁盘读取。
 *
 *	  此模块还必须处理从 WAL 中存储的单个块重新组装 toast 记录。
 *	  当一个元组的新（或初始）版本存储在 WAL 中时，它将始终有一些为
 *	  在行外存储的列发出的 toast 块在前。在单个顶级事务中，在行
 *	  的 toast 块和行数据之间将不会有其他承载数据的记录。有关详细
 *	  信息，请参见 ReorderBufferToast*。
 *
 *	  ReorderBuffer 使用两种特殊内存上下文类型 - SlabContext 用于
 *	  固定长度结构（更改和事务）的分配，GenerationContext 用于变
 *	  长事务数据（按相似生命周期的组分配和释放）。
 *
 *	  为了限制解码更改所使用的内存量，我们跟踪在重排序缓冲区级别
 *	  （即总内存量）和每个事务中使用的内存。当使用的内存总量超过
 *	  限制时，使用最多内存的事务将被序列化到磁盘。
 *
 *	  只有解码后的更改会从内存中驱逐（溢出到磁盘），而不是事务记录。
 *	  顶级事务的数量是有限的，但一个具有许多子事务的事务仍可能消耗
 *	  相当可观的内存。然而，事务记录相对较小，不包括在内存限制中。
 *
 *	  当前的驱逐算法非常简单 - 事务仅根据大小选择，而考虑更改的
 *	  年龄（LSN）可能是有用的。通过新的生成性内存分配器，驱逐最老的
 *	  更改将更有可能实际释放内存。
 *
 *	  在将序列化更改加载回内存时，我们仍然依赖于 max_changes_in_memory。
 *	  在那时我们无法直接使用内存限制，因为我们独立加载子事务。
 *	  处理此事的一种方法是计算子事务并允许每个子事务分配 1/N 的
 *	  内存限制。然而，这并不太吸引人，因为许多子事务可能很容易
 *	  导致抖动（反复解序列化和应用极少数更改的短周期）。我们可能
 *	  应该给予最老的子事务更多内存，因为它们可能为下一系列更改
 *	  提供来源。
 *
 * -------------------------------------------------------------------------
 */
#include "postgres.h"

#include <unistd.h>
#include <sys/stat.h>

#include "access/detoast.h"
#include "access/heapam.h"
#include "access/rewriteheap.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "catalog/catalog.h"
#include "lib/binaryheap.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/logical.h"
#include "replication/reorderbuffer.h"
#include "replication/slot.h"
#include "replication/snapbuild.h"	/* 仅用于 SnapBuildSnapDecRefcount */
#include "storage/bufmgr.h"
#include "storage/fd.h"
#include "storage/sinval.h"
#include "utils/builtins.h"
#include "utils/combocid.h"
#include "utils/memdebug.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/relfilenodemap.h"


/* 我们用于将 xid 映射到事务状态的哈希表项 */
typedef struct ReorderBufferTXNByIdEnt
{
	TransactionId xid;
	ReorderBufferTXN *txn;
} ReorderBufferTXNByIdEnt;

/* (relfilenode、ctid) => (cmin、cmax) 映射的数据结构 */
typedef struct ReorderBufferTupleCidKey
{
	RelFileNode relnode;
	ItemPointerData tid;
} ReorderBufferTupleCidKey;

typedef struct ReorderBufferTupleCidEnt
{
	ReorderBufferTupleCidKey key;
	CommandId	cmin;
	CommandId	cmax;
	CommandId	combocid;		/* 仅用于调试 */
} ReorderBufferTupleCidEnt;

/* 带有文件偏移跟踪的虚拟文件描述符 */
typedef struct TXNEntryFile
{
	File		vfd;			/* 文件关闭时为 -1 */
	off_t		curOffset;		/* 下一个写入或读取的偏移量。打开 vfd 时重置为 0 */
} TXNEntryFile;

/* k-way 有序变化迭代支持结构 */
typedef struct ReorderBufferIterTXNEntry
{
	XLogRecPtr	lsn;
	ReorderBufferChange *change;
	ReorderBufferTXN *txn;
	TXNEntryFile file;
	XLogSegNo	segno;
} ReorderBufferIterTXNEntry;

typedef struct ReorderBufferIterTXNState
{
	binaryheap *heap;
	Size		nr_txns;
	dlist_head	old_change;
	ReorderBufferIterTXNEntry entries[FLEXIBLE_ARRAY_MEMBER];
} ReorderBufferIterTXNState;

/* toast 数据结构 */
typedef struct ReorderBufferToastEnt
{
	Oid			chunk_id;		/* toast_table.chunk_id */
	int32		last_chunk_seq; /* 我们已看到的最后一个块的 toast_table.chunk_seq */
	Size		num_chunks;		/* 我们已经看到的块的数量 */
	Size		size;			/* 已看到的块的组合大小 */
	dlist_head	chunks;			/* 块的链表 */
	struct varlena *reconstructed;	/* 现在在主元组中指向重建的 varlena */
} ReorderBufferToastEnt;

/* 磁盘序列化支持数据结构 */
typedef struct ReorderBufferDiskChange
{
	Size		size;
	ReorderBufferChange change;
	/* 数据如下 */
} ReorderBufferDiskChange;

#define IsSpecInsert(action) \
( \
	((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT) \
)
#define IsSpecConfirmOrAbort(action) \
( \
	(((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM) || \
	((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT)) \
)
#define IsInsertOrUpdate(action) \
( \
	(((action) == REORDER_BUFFER_CHANGE_INSERT) || \
	((action) == REORDER_BUFFER_CHANGE_UPDATE) || \
	((action) == REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT)) \
)

/*
 * 每个事务在内存中保留的最大更改数量。之后，
 * 更改被转储到磁盘。
 *
 * 当前值应足以在 OLTP 工作负载下解码整个事务
 * 而无需访问磁盘，而在其他工作负载下合理快速地开始转储到磁盘。
 *
 * 在未来的某个时候，可能更合理地在这里进行更复杂的
 * 资源管理，但目前尚不清楚这将是什么样子。
 */
int			logical_decoding_work_mem;
static const Size max_changes_in_memory = 4096; /* XXX 仅用于恢复 */

/* ---------------------------------------
 * 主重排序缓冲区支持例程
 * ---------------------------------------
 */
static ReorderBufferTXN *fc_ReorderBufferGetTXN(ReorderBuffer *fc_rb);
static void fc_ReorderBufferReturnTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn);
static ReorderBufferTXN *fc_ReorderBufferTXNByXid(ReorderBuffer *fc_rb,
											   TransactionId fc_xid, bool fc_create, bool *fc_is_new,
											   XLogRecPtr fc_lsn, bool fc_create_as_top);
static void fc_ReorderBufferTransferSnapToParent(ReorderBufferTXN *fc_txn,
											  ReorderBufferTXN *fc_subtxn);

static void fc_AssertTXNLsnOrder(ReorderBuffer *fc_rb);

/* ---------------------------------------
 * 用于根据 lsn 顺序迭代事务及其子事务的 ->changes 的支持函数
 *
 * 用于迭代一个事务及其子事务的 k-way 堆合并
 * ---------------------------------------
 */
static void fc_ReorderBufferIterTXNInit(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
									 ReorderBufferIterTXNState *volatile *fc_iter_state);
static ReorderBufferChange *fc_ReorderBufferIterTXNNext(ReorderBuffer *fc_rb, ReorderBufferIterTXNState *fc_state);
static void fc_ReorderBufferIterTXNFinish(ReorderBuffer *fc_rb,
									   ReorderBufferIterTXNState *fc_state);
static void fc_ReorderBufferExecuteInvalidations(uint32 fc_nmsgs, SharedInvalidationMessage *fc_msgs);

/*
 * ---------------------------------------
 * 磁盘序列化支持函数
 * ---------------------------------------
 */
static void fc_ReorderBufferCheckMemoryLimit(ReorderBuffer *fc_rb);
static void fc_ReorderBufferSerializeTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn);
static void fc_ReorderBufferSerializeChange(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
										 int fc_fd, ReorderBufferChange *fc_change);
static Size fc_ReorderBufferRestoreChanges(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
										TXNEntryFile *fc_file, XLogSegNo *fc_segno);
static void fc_ReorderBufferRestoreChange(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
									   char *fc_change);
static void fc_ReorderBufferRestoreCleanup(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn);
static void fc_ReorderBufferTruncateTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
									 bool fc_txn_prepared);
static void fc_ReorderBufferCleanupSerializedTXNs(const char *fc_slotname);
static void fc_ReorderBufferSerializedPath(char *fc_path, ReplicationSlot *fc_slot,
										TransactionId fc_xid, XLogSegNo fc_segno);

static void fc_ReorderBufferFreeSnap(ReorderBuffer *fc_rb, Snapshot fc_snap);
static Snapshot fc_ReorderBufferCopySnap(ReorderBuffer *fc_rb, Snapshot fc_orig_snap,
									  ReorderBufferTXN *fc_txn, CommandId fc_cid);

/*
 * ---------------------------------------
 * 流式支持函数
 * ---------------------------------------
 */
static inline bool fc_ReorderBufferCanStream(ReorderBuffer *fc_rb);
static inline bool fc_ReorderBufferCanStartStreaming(ReorderBuffer *fc_rb);
static void fc_ReorderBufferStreamTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn);
static void fc_ReorderBufferStreamCommit(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn);

/* ---------------------------------------
 * toast 重新组装支持
 * ---------------------------------------
 */
static void fc_ReorderBufferToastInitHash(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn);
static void fc_ReorderBufferToastReset(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn);
static void fc_ReorderBufferToastReplace(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
									  Relation fc_relation, ReorderBufferChange *fc_change);
static void fc_ReorderBufferToastAppendChunk(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
										  Relation fc_relation, ReorderBufferChange *fc_change);

/*
 * ---------------------------------------
 * 内存计数
 * ---------------------------------------
 */
static Size fc_ReorderBufferChangeSize(ReorderBufferChange *fc_change);
static void fc_ReorderBufferChangeMemoryUpdate(ReorderBuffer *fc_rb,
											ReorderBufferChange *fc_change,
											bool fc_addition, Size fc_sz);

/*
 * 分配一个新的 ReorderBuffer，并清除来自
 * 同一槽的旧序列化状态。
 */
ReorderBuffer *
ReorderBufferAllocate(void)
{
	ReorderBuffer *fc_buffer;
	HASHCTL		fc_hash_ctl;
	MemoryContext fc_new_ctx;

	Assert(MyReplicationSlot != NULL);

	/* 在自己的上下文中分配内存，以便更好地进行核算 */
	fc_new_ctx = AllocSetContextCreate(CurrentMemoryContext,
									"ReorderBuffer",
									ALLOCSET_DEFAULT_SIZES);

	fc_buffer =
		(ReorderBuffer *) MemoryContextAlloc(fc_new_ctx, sizeof(ReorderBuffer));

	memset(&fc_hash_ctl, 0, sizeof(fc_hash_ctl));

	fc_buffer->context = fc_new_ctx;

	fc_buffer->change_context = SlabContextCreate(fc_new_ctx,
											   "Change",
											   SLAB_DEFAULT_BLOCK_SIZE,
											   sizeof(ReorderBufferChange));

	fc_buffer->txn_context = SlabContextCreate(fc_new_ctx,
											"TXN",
											SLAB_DEFAULT_BLOCK_SIZE,
											sizeof(ReorderBufferTXN));

	/*
	 * 为了减少由于跨多个内存块的长时间运行事务而导致的内存碎片，我们使用一个用于解码元组存储的固定大小内存块。性能测试表明，默认的内存块大小保持了逻辑解码性能，而不会由于并发事务而导致碎片化。有人可能会认为我们可以使用最大值作为SLAB_LARGE_BLOCK_SIZE，但测试也表明这无助于解决内存碎片问题。
	 */
	fc_buffer->tup_context = GenerationContextCreate(fc_new_ctx,
												  "Tuples",
												  SLAB_DEFAULT_BLOCK_SIZE,
												  SLAB_DEFAULT_BLOCK_SIZE,
												  SLAB_DEFAULT_BLOCK_SIZE);

	fc_hash_ctl.keysize = sizeof(TransactionId);
	fc_hash_ctl.entrysize = sizeof(ReorderBufferTXNByIdEnt);
	fc_hash_ctl.hcxt = fc_buffer->context;

	fc_buffer->by_txn = hash_create("ReorderBufferByXid", 1000, &fc_hash_ctl,
								 HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);

	fc_buffer->by_txn_last_xid = InvalidTransactionId;
	fc_buffer->by_txn_last_txn = NULL;

	fc_buffer->outbuf = NULL;
	fc_buffer->outbufsize = 0;
	fc_buffer->size = 0;

	fc_buffer->spillTxns = 0;
	fc_buffer->spillCount = 0;
	fc_buffer->spillBytes = 0;
	fc_buffer->streamTxns = 0;
	fc_buffer->streamCount = 0;
	fc_buffer->streamBytes = 0;
	fc_buffer->totalTxns = 0;
	fc_buffer->totalBytes = 0;

	fc_buffer->current_restart_decoding_lsn = InvalidXLogRecPtr;

	dlist_init(&fc_buffer->toplevel_by_lsn);
	dlist_init(&fc_buffer->txns_by_base_snapshot_lsn);

	/*
	 * 确保此槽位没有来自之前使用的过期数据，以防某些先前的退出避免调用ReorderBufferFree。未能做到这一点可能会产生重复的事务，同时如果没有数据则成本非常低。
	 */
	fc_ReorderBufferCleanupSerializedTXNs(NameStr(MyReplicationSlot->data.name));

	return fc_buffer;
}

/*
 * 释放一个ReorderBuffer
 */
void ReorderBufferFree(ReorderBuffer *fc_rb)
{
	MemoryContext fc_context = fc_rb->context;

	/*
	 * 我们通过完全废弃reorderbuffer的内存上下文来释放单独分配的数据。
	 */
	MemoryContextDelete(fc_context);

	/* 释放未使用的重排序缓冲区占用的磁盘空间 */
	fc_ReorderBufferCleanupSerializedTXNs(NameStr(MyReplicationSlot->data.name));
}

/*
 * 获取一个未使用的、可能预分配的ReorderBufferTXN。
 */
static ReorderBufferTXN * fc_ReorderBufferGetTXN(ReorderBuffer *fc_rb)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = (ReorderBufferTXN *)
		MemoryContextAlloc(fc_rb->txn_context, sizeof(ReorderBufferTXN));

	memset(fc_txn, 0, sizeof(ReorderBufferTXN));

	dlist_init(&fc_txn->changes);
	dlist_init(&fc_txn->tuplecids);
	dlist_init(&fc_txn->subtxns);

	/* InvalidCommandId不为零，因此显式地将其设置 */
	fc_txn->command_id = InvalidCommandId;
	fc_txn->output_plugin_private = NULL;

	return fc_txn;
}

/*
 * 释放一个ReorderBufferTXN。
 */
static void fc_ReorderBufferReturnTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	/* 如果我们被缓存（很可能），则清除查找缓存 */
	if (fc_rb->by_txn_last_xid == fc_txn->xid)
	{
		fc_rb->by_txn_last_xid = InvalidTransactionId;
		fc_rb->by_txn_last_txn = NULL;
	}

	/* 释放包含的数据 */

	if (fc_txn->gid != NULL)
	{
		pfree(fc_txn->gid);
		fc_txn->gid = NULL;
	}

	if (fc_txn->tuplecid_hash != NULL)
	{
		hash_destroy(fc_txn->tuplecid_hash);
		fc_txn->tuplecid_hash = NULL;
	}

	if (fc_txn->invalidations)
	{
		pfree(fc_txn->invalidations);
		fc_txn->invalidations = NULL;
	}

	/* 重置toast哈希 */
	fc_ReorderBufferToastReset(fc_rb, fc_txn);

	pfree(fc_txn);
}

/*
 * 获取一个新的ReorderBufferChange。
 */
ReorderBufferChange *
ReorderBufferGetChange(ReorderBuffer *fc_rb)
{
	ReorderBufferChange *fc_change;

	fc_change = (ReorderBufferChange *)
		MemoryContextAlloc(fc_rb->change_context, sizeof(ReorderBufferChange));

	memset(fc_change, 0, sizeof(ReorderBufferChange));
	return fc_change;
}

/*
 * 释放一个ReorderBufferChange，并在请求时更新内存帐户。
 */
void ReorderBufferReturnChange(ReorderBuffer *fc_rb, ReorderBufferChange *fc_change,
						  bool fc_upd_mem)
{
	/* 更新内存帐户信息 */
	if (fc_upd_mem)
		fc_ReorderBufferChangeMemoryUpdate(fc_rb, fc_change, false,
										fc_ReorderBufferChangeSize(fc_change));

	/* 释放包含的数据 */
	switch (fc_change->action)
	{
		case REORDER_BUFFER_CHANGE_INSERT:
		case REORDER_BUFFER_CHANGE_UPDATE:
		case REORDER_BUFFER_CHANGE_DELETE:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
			if (fc_change->data.tp.newtuple)
			{
				ReorderBufferReturnTupleBuf(fc_rb, fc_change->data.tp.newtuple);
				fc_change->data.tp.newtuple = NULL;
			}

			if (fc_change->data.tp.oldtuple)
			{
				ReorderBufferReturnTupleBuf(fc_rb, fc_change->data.tp.oldtuple);
				fc_change->data.tp.oldtuple = NULL;
			}
			break;
		case REORDER_BUFFER_CHANGE_MESSAGE:
			if (fc_change->data.msg.prefix != NULL)
				pfree(fc_change->data.msg.prefix);
			fc_change->data.msg.prefix = NULL;
			if (fc_change->data.msg.message != NULL)
				pfree(fc_change->data.msg.message);
			fc_change->data.msg.message = NULL;
			break;
		case REORDER_BUFFER_CHANGE_INVALIDATION:
			if (fc_change->data.inval.invalidations)
				pfree(fc_change->data.inval.invalidations);
			fc_change->data.inval.invalidations = NULL;
			break;
		case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
			if (fc_change->data.snapshot)
			{
				fc_ReorderBufferFreeSnap(fc_rb, fc_change->data.snapshot);
				fc_change->data.snapshot = NULL;
			}
			break;
			/* 除了结构本身外没有其他数据 */
		case REORDER_BUFFER_CHANGE_TRUNCATE:
			if (fc_change->data.truncate.relids != NULL)
			{
				ReorderBufferReturnRelids(fc_rb, fc_change->data.truncate.relids);
				fc_change->data.truncate.relids = NULL;
			}
			break;
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
			break;
	}

	pfree(fc_change);
}

/*
 * 获取一个新的ReorderBufferTupleBuf，至少适合大小为tuple_len的元组
 *（不包括头部开销）。
 */
ReorderBufferTupleBuf *
ReorderBufferGetTupleBuf(ReorderBuffer *fc_rb, Size fc_tuple_len)
{
	ReorderBufferTupleBuf *fc_tuple;
	Size		fc_alloc_len;

	fc_alloc_len = fc_tuple_len + SizeofHeapTupleHeader;

	fc_tuple = (ReorderBufferTupleBuf *)
		MemoryContextAlloc(fc_rb->tup_context,
						   sizeof(ReorderBufferTupleBuf) +
						   MAXIMUM_ALIGNOF + fc_alloc_len);
	fc_tuple->alloc_tuple_size = fc_alloc_len;
	fc_tuple->tuple.t_data = ReorderBufferTupleBufData(fc_tuple);

	return fc_tuple;
}

/*
 * 释放一个ReorderBufferTupleBuf。
 */
void ReorderBufferReturnTupleBuf(ReorderBuffer *fc_rb, ReorderBufferTupleBuf *fc_tuple)
{
	pfree(fc_tuple);
}

/*
 * 获取一个用于截断关系的relids数组。
 *
 * 我们使用全局内存上下文（用于整个重排序缓冲区），因为现有的上下文似乎没有一个合适（有些是SLAB，因此我们不能使用它们，tup_context是用于元组数据，而不是relids）。我们可以再添加一个上下文，但这似乎有些过剩 - TRUNCATE并不是特别常见的操作，因此似乎不值得。
 */
Oid * ReorderBufferGetRelids(ReorderBuffer *fc_rb, int fc_nrelids)
{
	Oid		   *fc_relids;
	Size		fc_alloc_len;

	fc_alloc_len = sizeof(Oid) * fc_nrelids;

	fc_relids = (Oid *) MemoryContextAlloc(fc_rb->context, fc_alloc_len);

	return fc_relids;
}

/*
 * 释放relids数组。
 */
void ReorderBufferReturnRelids(ReorderBuffer *fc_rb, Oid *fc_relids)
{
	pfree(fc_relids);
}

/*
 * 从给定的缓冲区返回ReorderBufferTXN，按Xid指定。
 * 如果create为真，并且事务尚不存在，则创建它
 * （使用给定的LSN，并作为顶级事务如果指定的话）；当发生这种情况时，is_new被设置为true。
 */
static ReorderBufferTXN * fc_ReorderBufferTXNByXid(ReorderBuffer *fc_rb, TransactionId fc_xid, bool fc_create,
					  bool *fc_is_new, XLogRecPtr fc_lsn, bool fc_create_as_top)
{
	ReorderBufferTXN *fc_txn;
	ReorderBufferTXNByIdEnt *fc_ent;
	bool		fc_found;

	Assert(TransactionIdIsValid(fc_xid));

	/*
	 * 首先检查一个条目的查找缓存
	 */
	if (TransactionIdIsValid(fc_rb->by_txn_last_xid) &&
		fc_rb->by_txn_last_xid == fc_xid)
	{
		fc_txn = fc_rb->by_txn_last_txn;

		if (fc_txn != NULL)
		{
			/* 找到了，并且它是有效的 */
			if (fc_is_new)
				*fc_is_new = false;
			return fc_txn;
		}

		/*
		 * 被缓存为不存在，并要求不创建？那么没有其他要做的了。
		 */
		if (!fc_create)
			return NULL;
		/* 否则则继续创建它 */
	}

	/*
	 * 如果缓存未命中或返回了“不存在”，并且我们想要
	 * 创建一个条目。
	 */

	/* 搜索查找表 */
	fc_ent = (ReorderBufferTXNByIdEnt *)
		hash_search(fc_rb->by_txn,
					(void *) &fc_xid,
					fc_create ? HASH_ENTER : HASH_FIND,
					&fc_found);
	if (fc_found)
		fc_txn = fc_ent->txn;
	else if (fc_create)
	{
		/* 如果请求创建，则初始化新条目 */
		Assert(fc_ent != NULL);
		Assert(fc_lsn != InvalidXLogRecPtr);

		fc_ent->txn = fc_ReorderBufferGetTXN(fc_rb);
		fc_ent->txn->xid = fc_xid;
		fc_txn = fc_ent->txn;
		fc_txn->first_lsn = fc_lsn;
		fc_txn->restart_decoding_lsn = fc_rb->current_restart_decoding_lsn;

		if (fc_create_as_top)
		{
			dlist_push_tail(&fc_rb->toplevel_by_lsn, &fc_txn->node);
			fc_AssertTXNLsnOrder(fc_rb);
		}
	}
	else
		fc_txn = NULL;				/* 未找到且未请求创建 */

	/* 更新缓存 */
	fc_rb->by_txn_last_xid = fc_xid;
	fc_rb->by_txn_last_txn = fc_txn;

	if (fc_is_new)
		*fc_is_new = !fc_found;

	Assert(!fc_create || fc_txn != NULL);
	return fc_txn;
}

/*
 * 记录正在进行的事务流的部分更改。
 * 我们只能流式传输完整的更改，因此如果我们有像 toast
 * 表插入或猜测插入这样的部分更改，则将此类“txn”标记为
 * 不能流式传输。我们还确保，如果此类“txn”中的更改
 * 超过 logical_decoding_work_mem 阈值，则在我们得到完整
 * 更改后，尽快流式传输它们。
 */
static void fc_ReorderBufferProcessPartialChange(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
								  ReorderBufferChange *fc_change,
								  bool fc_toast_insert)
{
	ReorderBufferTXN *fc_toptxn;

	/*
	 * 仅在流式传输进行中的事务时处理部分更改。
	 */
	if (!fc_ReorderBufferCanStream(fc_rb))
		return;

	/* 获取顶级事务。 */
	if (fc_txn->toptxn != NULL)
		fc_toptxn = fc_txn->toptxn;
	else
		fc_toptxn = fc_txn;

	/*
	 * 表示 toast 插入的部分更改。一旦我们获得主表
	 * 的插入或更新，并且我们确定待处理的 toast 块不再
	 * 被需要，该更改将被视为完整。
	 *
	 * 如果我们允许在有待处理的 toast 块时进行流式传输，
	 * 则这些块将在插入（multi_insert）完成之前不会被释放，
	 * 我们预计 txn 会在流式传输之后流式传输所有更改。
	 * 这一限制主要是为了确保流式事务的正确性，似乎没有必要
	 * 仅仅为了允许这种情况而提升此限制，因为无论如何我们
	 * 都将在该插入完成后流式传输事务。
	 */
	if (fc_toast_insert)
		fc_toptxn->txn_flags |= RBTXN_HAS_PARTIAL_CHANGE;
	else if (rbtxn_has_partial_change(fc_toptxn) &&
			 IsInsertOrUpdate(fc_change->action) &&
			 fc_change->data.tp.clear_toast_afterwards)
		fc_toptxn->txn_flags &= ~RBTXN_HAS_PARTIAL_CHANGE;

	/*
	 * 表示猜测插入的部分更改。一旦我们获得猜测确认或
	 * 中止标记，该更改将被视为完整。
	 */
	if (IsSpecInsert(fc_change->action))
		fc_toptxn->txn_flags |= RBTXN_HAS_PARTIAL_CHANGE;
	else if (rbtxn_has_partial_change(fc_toptxn) &&
			 IsSpecConfirmOrAbort(fc_change->action))
		fc_toptxn->txn_flags &= ~RBTXN_HAS_PARTIAL_CHANGE;

	/*
	 * 如果它在之前被序列化且顶级事务中的更改现在已完整，
	 * 则流式传输该事务。
	 *
	 * 尽快流式传输此类事务的原因是，因为之前它将达到
	 * 内存阈值并且由于更改不完整而不能被流式传输。
	 * 延迟此类事务将为它们增加应用延迟。
	 */
	if (fc_ReorderBufferCanStartStreaming(fc_rb) &&
		!(rbtxn_has_partial_change(fc_toptxn)) &&
		rbtxn_is_serialized(fc_txn))
		fc_ReorderBufferStreamTXN(fc_rb, fc_toptxn);
}

/*
 * 将更改排入事务，以便在提交时重播或在达到
 * logical_decoding_work_mem 阈值时流式传输。
 */
void ReorderBufferQueueChange(ReorderBuffer *fc_rb, TransactionId fc_xid, XLogRecPtr fc_lsn,
						 ReorderBufferChange *fc_change, bool fc_toast_insert)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, true, NULL, fc_lsn, true);

	/*
	 * 在流式传输先前更改时，我们检测到事务已中止。
	 * 因此，收集更多更改是没有意义的。
	 */
	if (fc_txn->concurrent_abort)
	{
		/*
		 * 我们不需要更新此更改的内存账目，因为我们
		 * 还没有将其添加到队列。
		 */
		ReorderBufferReturnChange(fc_rb, fc_change, false);
		return;
	}

	fc_change->lsn = fc_lsn;
	fc_change->txn = fc_txn;

	Assert(InvalidXLogRecPtr != fc_lsn);
	dlist_push_tail(&fc_txn->changes, &fc_change->node);
	fc_txn->nentries++;
	fc_txn->nentries_mem++;

	/* 更新内存账目信息 */
	fc_ReorderBufferChangeMemoryUpdate(fc_rb, fc_change, true,
									fc_ReorderBufferChangeSize(fc_change));

	/* 处理部分更改 */
	fc_ReorderBufferProcessPartialChange(fc_rb, fc_txn, fc_change, fc_toast_insert);

	/* 检查内存限制，必要时驱逐某些东西 */
	fc_ReorderBufferCheckMemoryLimit(fc_rb);
}

/*
 * 事务消息被排队以在提交时处理，而非事务消息立即
 * 被处理。
 */
void ReorderBufferQueueMessage(ReorderBuffer *fc_rb, TransactionId fc_xid,
						  Snapshot fc_snapshot, XLogRecPtr fc_lsn,
						  bool fc_transactional, const char *fc_prefix,
						  Size fc_message_size, const char *fc_message)
{
	if (fc_transactional)
	{
		MemoryContext fc_oldcontext;
		ReorderBufferChange *fc_change;

		Assert(fc_xid != InvalidTransactionId);

		/*
		 * 我们不期望事务更改的快照 - 我们将在应用过程中
		 * 使用稍后派生的快照（除非该更改被跳过）。
		 */
		Assert(!fc_snapshot);

		fc_oldcontext = MemoryContextSwitchTo(fc_rb->context);

		fc_change = ReorderBufferGetChange(fc_rb);
		fc_change->action = REORDER_BUFFER_CHANGE_MESSAGE;
		fc_change->data.msg.prefix = pstrdup(fc_prefix);
		fc_change->data.msg.message_size = fc_message_size;
		fc_change->data.msg.message = palloc(fc_message_size);
		memcpy(fc_change->data.msg.message, fc_message, fc_message_size);

		ReorderBufferQueueChange(fc_rb, fc_xid, fc_lsn, fc_change, false);

		MemoryContextSwitchTo(fc_oldcontext);
	}
	else
	{
		ReorderBufferTXN *fc_txn = NULL;
		volatile Snapshot fc_snapshot_now = fc_snapshot;

		/* 非事务性更改需要有效的快照。 */
		Assert(fc_snapshot_now);

		if (fc_xid != InvalidTransactionId)
			fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, true, NULL, fc_lsn, true);

		/* 设置快照以允许访问目录 */
		SetupHistoricSnapshot(fc_snapshot_now, NULL);
		PG_TRY();
		{
			fc_rb->message(fc_rb, fc_txn, fc_lsn, false, fc_prefix, fc_message_size, fc_message);

			TeardownHistoricSnapshot(false);
		}
		PG_CATCH();
		{
			TeardownHistoricSnapshot(true);
			PG_RE_THROW();
		}
		PG_END_TRY();
	}
}

/*
 * AssertTXNLsnOrder
 *		验证重排序缓冲区中事务列表的 LSN 排序
 *
 * 还会检查其他与 LSN 相关的不变性。
 *
 * 如果不使用断言，则不操作。
 */
static void fc_AssertTXNLsnOrder(ReorderBuffer *fc_rb)
{
#ifdef USE_ASSERT_CHECKING
	LogicalDecodingContext *fc_ctx = fc_rb->private_data;
	dlist_iter	fc_iter;
	XLogRecPtr	fc_prev_first_lsn = InvalidXLogRecPtr;
	XLogRecPtr	fc_prev_base_snap_lsn = InvalidXLogRecPtr;

	/*
	 * 如果我们尚未达到开始解码事务内容的 LSN，则跳过验证，
	 * 因为在我们到达 LSN 之前，可能会有事务没有
	 * 顶层事务和子事务之间的关联，因此具有
	 * 相同的 LSN。我们并不保证此关联，直到我们尝试
	 * 解码实际的事务内容。在 start_decoding_at LSN 之前的记录
	 * 的顺序应在重启之前检查。
	 */
	if (SnapBuildXactNeedsSkip(fc_ctx->snapshot_builder, fc_ctx->reader->EndRecPtr))
		return;

	dlist_foreach(fc_iter, &fc_rb->toplevel_by_lsn)
	{
		ReorderBufferTXN *fc_cur_txn = dlist_container(ReorderBufferTXN, node,
													fc_iter.cur);

		/* 必须设置开始 LSN */
		Assert(fc_cur_txn->first_lsn != InvalidXLogRecPtr);

		/* 如果存在结束 LSN，必须高于开始 LSN */
		if (fc_cur_txn->end_lsn != InvalidXLogRecPtr)
			Assert(fc_cur_txn->first_lsn <= fc_cur_txn->end_lsn);

		/* 当前的初始 LSN 必须严格高于之前的 */
		if (fc_prev_first_lsn != InvalidXLogRecPtr)
			Assert(fc_prev_first_lsn < fc_cur_txn->first_lsn);

		/* 已知为子事务的 txns 不得被列出 */
		Assert(!rbtxn_is_known_subxact(fc_cur_txn));

		fc_prev_first_lsn = fc_cur_txn->first_lsn;
	}

	dlist_foreach(fc_iter, &fc_rb->txns_by_base_snapshot_lsn)
	{
		ReorderBufferTXN *fc_cur_txn = dlist_container(ReorderBufferTXN,
													base_snapshot_node,
													fc_iter.cur);

		/* 基础快照（及其 LSN）必须设置 */
		Assert(fc_cur_txn->base_snapshot != NULL);
		Assert(fc_cur_txn->base_snapshot_lsn != InvalidXLogRecPtr);

		/* 当前 LSN 必须严格高于之前的 */
		if (fc_prev_base_snap_lsn != InvalidXLogRecPtr)
			Assert(fc_prev_base_snap_lsn < fc_cur_txn->base_snapshot_lsn);

		/* 已知为子事务的 txns 不得被列出 */
		Assert(!rbtxn_is_known_subxact(fc_cur_txn));

		fc_prev_base_snap_lsn = fc_cur_txn->base_snapshot_lsn;
	}
#endif
}

/*
 * AssertChangeLsnOrder
 *
 * 检查（子）事务中变更的排序。
 */
static void fc_AssertChangeLsnOrder(ReorderBufferTXN *fc_txn)
{
#ifdef USE_ASSERT_CHECKING
	dlist_iter	fc_iter;
	XLogRecPtr	fc_prev_lsn = fc_txn->first_lsn;

	dlist_foreach(fc_iter, &fc_txn->changes)
	{
		ReorderBufferChange *fc_cur_change;

		fc_cur_change = dlist_container(ReorderBufferChange, node, fc_iter.cur);

		Assert(fc_txn->first_lsn != InvalidXLogRecPtr);
		Assert(fc_cur_change->lsn != InvalidXLogRecPtr);
		Assert(fc_txn->first_lsn <= fc_cur_change->lsn);

		if (fc_txn->end_lsn != InvalidXLogRecPtr)
			Assert(fc_cur_change->lsn <= fc_txn->end_lsn);

		Assert(fc_prev_lsn <= fc_cur_change->lsn);

		fc_prev_lsn = fc_cur_change->lsn;
	}
#endif
}

/*
 * ReorderBufferGetOldestTXN
 *		返回重排序缓冲区中的最旧事务
 */
ReorderBufferTXN *
ReorderBufferGetOldestTXN(ReorderBuffer *fc_rb)
{
	ReorderBufferTXN *fc_txn;

	fc_AssertTXNLsnOrder(fc_rb);

	if (dlist_is_empty(&fc_rb->toplevel_by_lsn))
		return NULL;

	fc_txn = dlist_head_element(ReorderBufferTXN, node, &fc_rb->toplevel_by_lsn);

	Assert(!rbtxn_is_known_subxact(fc_txn));
	Assert(fc_txn->first_lsn != InvalidXLogRecPtr);
	return fc_txn;
}

/*
 * ReorderBufferGetOldestXmin
 *		返回重排序缓冲区中的最旧 Xmin
 *
 * 从重排序缓冲区中保留的事务所用快照的角度返回可能正在运行的最旧 Xid，
 * 如果没有则返回 InvalidTransactionId。
 *
 * 由于快照是单调分配的，这等于具有最小 base_snapshot_lsn 的
 * 基础快照的 Xmin。
 */
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *fc_rb)
{
	ReorderBufferTXN *fc_txn;

	fc_AssertTXNLsnOrder(fc_rb);

	if (dlist_is_empty(&fc_rb->txns_by_base_snapshot_lsn))
		return InvalidTransactionId;

	fc_txn = dlist_head_element(ReorderBufferTXN, base_snapshot_node,
							 &fc_rb->txns_by_base_snapshot_lsn);
	return fc_txn->base_snapshot->xmin;
}

void ReorderBufferSetRestartPoint(ReorderBuffer *fc_rb, XLogRecPtr fc_ptr)
{
	fc_rb->current_restart_decoding_lsn = fc_ptr;
}

/*
 * ReorderBufferAssignChild
 *
 * 注明我们知道 subxid 是 xid 的子事务，以给定的 lsn 为准。
 */
void ReorderBufferAssignChild(ReorderBuffer *fc_rb, TransactionId fc_xid,
						 TransactionId fc_subxid, XLogRecPtr fc_lsn)
{
	ReorderBufferTXN *fc_txn;
	ReorderBufferTXN *fc_subtxn;
	bool		fc_new_top;
	bool		fc_new_sub;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, true, &fc_new_top, fc_lsn, true);
	fc_subtxn = fc_ReorderBufferTXNByXid(fc_rb, fc_subxid, true, &fc_new_sub, fc_lsn, false);

	if (!fc_new_sub)
	{
		if (rbtxn_is_known_subxact(fc_subtxn))
		{
			/* 已经关联，无需操作 */
			return;
		}
		else
		{
			/*
			 * 我们已经看到了这个事务，但最初将其添加到
			 * 顶层事务的列表中。现在我们知道它不是顶层事务，
			 * 从列表中将其移除。
			 */
			dlist_delete(&fc_subtxn->node);
		}
	}

	fc_subtxn->txn_flags |= RBTXN_IS_SUBXACT;
	fc_subtxn->toplevel_xid = fc_xid;
	Assert(fc_subtxn->nsubtxns == 0);

	/* 设置对顶层事务的引用 */
	fc_subtxn->toptxn = fc_txn;

	/* 添加到子事务列表 */
	dlist_push_tail(&fc_txn->subtxns, &fc_subtxn->node);
	fc_txn->nsubtxns++;

	/* 可能将子事务的快照转移到其顶层事务中。 */
	fc_ReorderBufferTransferSnapToParent(fc_txn, fc_subtxn);

	/* 验证 LSN 排序不变性 */
	fc_AssertTXNLsnOrder(fc_rb);
}

/*
 * ReorderBufferTransferSnapToParent
 *		如果需要，将基础快照从子事务传递到顶层事务
 *
 * 当顶层事务没有基础快照，或者子事务的基础快照的 LSN 比顶层事务的基础快照的 LSN 更早时，将执行此操作。 如果顶层事务没有更改，但子事务有一些更改，或者子事务中的第一个更改的 LSN 比顶层事务中的第一个更改的 LSN 更早，并且我们刚刚了解到它们的亲属关系，这种情况也可能发生。
 *
 * 无论是否发生转移，子事务的快照都会被清除，因为无论哪种情况，它都不再需要。
 *
 * 我们一旦意识到它们的亲属关系，就会立即执行此操作，以避免将额外的快照排队到已知为子事务的事务中 -- 只有顶层事务将接收进一步的快照。
 */
static void fc_ReorderBufferTransferSnapToParent(ReorderBufferTXN *fc_txn,
								  ReorderBufferTXN *fc_subtxn)
{
	Assert(fc_subtxn->toplevel_xid == fc_txn->xid);

	if (fc_subtxn->base_snapshot != NULL)
	{
		if (fc_txn->base_snapshot == NULL ||
			fc_subtxn->base_snapshot_lsn < fc_txn->base_snapshot_lsn)
		{
			/*
			 * 如果顶层事务已经有基础快照，但它比子事务的快照更新，则清除它。
			 */
			if (fc_txn->base_snapshot != NULL)
			{
				SnapBuildSnapDecRefcount(fc_txn->base_snapshot);
				dlist_delete(&fc_txn->base_snapshot_node);
			}

			/*
			 * 现在快照是顶层事务的；将其转移，并通过将其移动到子事务所在的位置来调整顶层事务在列表中的位置。
			 */
			fc_txn->base_snapshot = fc_subtxn->base_snapshot;
			fc_txn->base_snapshot_lsn = fc_subtxn->base_snapshot_lsn;
			dlist_insert_before(&fc_subtxn->base_snapshot_node,
								&fc_txn->base_snapshot_node);

			/*
			 * 子事务不再有快照（因此它不会在列表中）。
			 */
			fc_subtxn->base_snapshot = NULL;
			fc_subtxn->base_snapshot_lsn = InvalidXLogRecPtr;
			dlist_delete(&fc_subtxn->base_snapshot_node);
		}
		else
		{
			/* 顶层事务的基础快照正常，因此子事务的快照不再需要 */
			SnapBuildSnapDecRefcount(fc_subtxn->base_snapshot);
			dlist_delete(&fc_subtxn->base_snapshot_node);
			fc_subtxn->base_snapshot = NULL;
			fc_subtxn->base_snapshot_lsn = InvalidXLogRecPtr;
		}
	}
}

/*
 * 在提交时将子事务与其顶层事务关联。 此后可能不会添加进一步的更改。
 */
void ReorderBufferCommitChild(ReorderBuffer *fc_rb, TransactionId fc_xid,
						 TransactionId fc_subxid, XLogRecPtr fc_commit_lsn,
						 XLogRecPtr fc_end_lsn)
{
	ReorderBufferTXN *fc_subtxn;

	fc_subtxn = fc_ReorderBufferTXNByXid(fc_rb, fc_subxid, false, NULL,
								   InvalidXLogRecPtr, false);

	/*
	 * 如果该子事务没有包含任何更改，则无需执行任何操作
	 */
	if (!fc_subtxn)
		return;

	fc_subtxn->final_lsn = fc_commit_lsn;
	fc_subtxn->end_lsn = fc_end_lsn;

	/*
	 * 将此子事务分配为顶层事务的子事务（如果已经完成则无操作）。
	 */
	ReorderBufferAssignChild(fc_rb, fc_xid, fc_subxid, InvalidXLogRecPtr);
}


/*
 * 支持高效遍历事务及其子事务的更改。
 *
 * 我们通过在事务/子事务之间进行 k-路合并来实现。 为此，我们将不同事务的当前头部建模为二进制堆，以便轻松知道哪个（子）事务下一个有最小的 lsn 的更改。
 *
 * 我们假设单个事务中的更改已经按 LSN 排序。
 */

/*
 * 二进制堆比较函数。
 */
static int fc_ReorderBufferIterCompare(Datum fc_a, Datum fc_b, void *fc_arg)
{
	ReorderBufferIterTXNState *fc_state = (ReorderBufferIterTXNState *) fc_arg;
	XLogRecPtr	fc_pos_a = fc_state->entries[DatumGetInt32(fc_a)].lsn;
	XLogRecPtr	fc_pos_b = fc_state->entries[DatumGetInt32(fc_b)].lsn;

	if (fc_pos_a < fc_pos_b)
		return 1;
	else if (fc_pos_a == fc_pos_b)
		return 0;
	return -1;
}

/*
 * 分配并初始化一个迭代器，该迭代器按 LSN 顺序遍历一个事务及其所有子事务。
 *
 * 注意：迭代器状态通过 iter_state 参数返回，而不是函数的返回值。 这是因为状态在调用者的 PG_CATCH 块中被清理，因此我们希望确保即使此函数抛出异常，调用者也能返回状态。
 */
static void fc_ReorderBufferIterTXNInit(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
						 ReorderBufferIterTXNState *volatile *fc_iter_state)
{
	Size		fc_nr_txns = 0;
	ReorderBufferIterTXNState *fc_state;
	dlist_iter	fc_cur_txn_i;
	int32		fc_off;

	*fc_iter_state = NULL;

	/* 检查顶层事务中更改的顺序。 */
	fc_AssertChangeLsnOrder(fc_txn);

	/*
	 * 计算我们的堆的大小：每个包含更改的事务一个元素。 （除了已经在重新排序缓冲区中的事务外，我们还计算直接传递给我们的那个。）
	 */
	if (fc_txn->nentries > 0)
		fc_nr_txns++;

	dlist_foreach(fc_cur_txn_i, &fc_txn->subtxns)
	{
		ReorderBufferTXN *fc_cur_txn;

		fc_cur_txn = dlist_container(ReorderBufferTXN, node, fc_cur_txn_i.cur);

		/* 检查此子事务中更改的顺序。 */
		fc_AssertChangeLsnOrder(fc_cur_txn);

		if (fc_cur_txn->nentries > 0)
			fc_nr_txns++;
	}

	/* 分配迭代状态 */
	fc_state = (ReorderBufferIterTXNState *)
		MemoryContextAllocZero(fc_rb->context,
							   sizeof(ReorderBufferIterTXNState) +
							   sizeof(ReorderBufferIterTXNEntry) * fc_nr_txns);

	fc_state->nr_txns = fc_nr_txns;
	dlist_init(&fc_state->old_change);

	for (fc_off = 0; fc_off < fc_state->nr_txns; fc_off++)
	{
		fc_state->entries[fc_off].file.vfd = -1;
		fc_state->entries[fc_off].segno = 0;
	}

	/* 分配堆 */
	fc_state->heap = binaryheap_allocate(fc_state->nr_txns,
									  fc_ReorderBufferIterCompare,
									  fc_state);

	/* 现在状态字段已初始化，可以安全返回。 */
	*fc_iter_state = fc_state;

	/*
	 * 现在以无序的方式将项目插入二叉堆。（我们将在最后执行堆组装步骤；这样更高效。）
	 */

	fc_off = 0;

	/* 如果顶级事务包含更改，则添加顶级事务 */
	if (fc_txn->nentries > 0)
	{
		ReorderBufferChange *fc_cur_change;

		if (rbtxn_is_serialized(fc_txn))
		{
			/* 序列化剩余更改 */
			fc_ReorderBufferSerializeTXN(fc_rb, fc_txn);
			fc_ReorderBufferRestoreChanges(fc_rb, fc_txn, &fc_state->entries[fc_off].file,
										&fc_state->entries[fc_off].segno);
		}

		fc_cur_change = dlist_head_element(ReorderBufferChange, node,
										&fc_txn->changes);

		fc_state->entries[fc_off].lsn = fc_cur_change->lsn;
		fc_state->entries[fc_off].change = fc_cur_change;
		fc_state->entries[fc_off].txn = fc_txn;

		binaryheap_add_unordered(fc_state->heap, Int32GetDatum(fc_off++));
	}

	/* 如果子事务包含更改，则添加子事务 */
	dlist_foreach(fc_cur_txn_i, &fc_txn->subtxns)
	{
		ReorderBufferTXN *fc_cur_txn;

		fc_cur_txn = dlist_container(ReorderBufferTXN, node, fc_cur_txn_i.cur);

		if (fc_cur_txn->nentries > 0)
		{
			ReorderBufferChange *fc_cur_change;

			if (rbtxn_is_serialized(fc_cur_txn))
			{
				/* 序列化剩余更改 */
				fc_ReorderBufferSerializeTXN(fc_rb, fc_cur_txn);
				fc_ReorderBufferRestoreChanges(fc_rb, fc_cur_txn,
											&fc_state->entries[fc_off].file,
											&fc_state->entries[fc_off].segno);
			}
			fc_cur_change = dlist_head_element(ReorderBufferChange, node,
											&fc_cur_txn->changes);

			fc_state->entries[fc_off].lsn = fc_cur_change->lsn;
			fc_state->entries[fc_off].change = fc_cur_change;
			fc_state->entries[fc_off].txn = fc_cur_txn;

			binaryheap_add_unordered(fc_state->heap, Int32GetDatum(fc_off++));
		}
	}

	/* 组装一个有效的二叉堆 */
	binaryheap_build(fc_state->heap);
}

/*
 * 在迭代事务及其子事务时返回下一个更改。
 *
 * 当没有更多更改时返回NULL。
 */
static ReorderBufferChange *
fc_ReorderBufferIterTXNNext(ReorderBuffer *fc_rb, ReorderBufferIterTXNState *fc_state)
{
	ReorderBufferChange *fc_change;
	ReorderBufferIterTXNEntry *fc_entry;
	int32		fc_off;

	/* 这里没有东西了 */
	if (fc_state->heap->bh_size == 0)
		return NULL;

	fc_off = DatumGetInt32(binaryheap_first(fc_state->heap));
	fc_entry = &fc_state->entries[fc_off];

	/* 释放我们在上一个*Next调用中可能"泄漏"的内存 */
	if (!dlist_is_empty(&fc_state->old_change))
	{
		fc_change = dlist_container(ReorderBufferChange, node,
								 dlist_pop_head_node(&fc_state->old_change));
		ReorderBufferReturnChange(fc_rb, fc_change, true);
		Assert(dlist_is_empty(&fc_state->old_change));
	}

	fc_change = fc_entry->change;

	/*
	 * 使用有关哪个事务在LSN顺序中具有下一个相关更改的信息更新堆
	 */

	/* 内存中有更改 */
	if (dlist_has_next(&fc_entry->txn->changes, &fc_entry->change->node))
	{
		dlist_node *fc_next = dlist_next_node(&fc_entry->txn->changes, &fc_change->node);
		ReorderBufferChange *fc_next_change =
		dlist_container(ReorderBufferChange, node, fc_next);

		/* txn保持不变 */
		fc_state->entries[fc_off].lsn = fc_next_change->lsn;
		fc_state->entries[fc_off].change = fc_next_change;

		binaryheap_replace_first(fc_state->heap, Int32GetDatum(fc_off));
		return fc_change;
	}

	/* 尝试从磁盘加载更改 */
	if (fc_entry->txn->nentries != fc_entry->txn->nentries_mem)
	{
		/*
		 * 糟糕：还原更改将重用*Change记录，因此从每个事务的列表中删除当前记录，并仅在下一个调用中释放。
		 */
		dlist_delete(&fc_change->node);
		dlist_push_tail(&fc_state->old_change, &fc_change->node);

		/*
		 * 更新我们正在释放当前更改集并还原新更改集的事务处理的总字节数。
		 */
		fc_rb->totalBytes += fc_entry->txn->size;
		if (fc_ReorderBufferRestoreChanges(fc_rb, fc_entry->txn, &fc_entry->file,
										&fc_state->entries[fc_off].segno))
		{
			/* 成功从磁盘还原更改 */
			ReorderBufferChange *fc_next_change =
			dlist_head_element(ReorderBufferChange, node,
							   &fc_entry->txn->changes);

			elog(DEBUG2, "restored %u/%u changes from disk",
				 (uint32) fc_entry->txn->nentries_mem,
				 (uint32) fc_entry->txn->nentries);

			Assert(fc_entry->txn->nentries_mem);
			/* txn保持不变 */
			fc_state->entries[fc_off].lsn = fc_next_change->lsn;
			fc_state->entries[fc_off].change = fc_next_change;
			binaryheap_replace_first(fc_state->heap, Int32GetDatum(fc_off));

			return fc_change;
		}
	}

	/* 好吧，那里没有更改了，移除 */
	binaryheap_remove_first(fc_state->heap);

	return fc_change;
}

/*
 * 释放迭代器
 */
static void fc_ReorderBufferIterTXNFinish(ReorderBuffer *fc_rb,
						   ReorderBufferIterTXNState *fc_state)
{
	int32		fc_off;

	for (fc_off = 0; fc_off < fc_state->nr_txns; fc_off++)
	{
		if (fc_state->entries[fc_off].file.vfd != -1)
			FileClose(fc_state->entries[fc_off].file.vfd);
	}

	/* 释放我们在上一个*Next调用中可能"泄漏"的内存 */
	if (!dlist_is_empty(&fc_state->old_change))
	{
		ReorderBufferChange *fc_change;

		fc_change = dlist_container(ReorderBufferChange, node,
								 dlist_pop_head_node(&fc_state->old_change));
		ReorderBufferReturnChange(fc_rb, fc_change, true);
		Assert(dlist_is_empty(&fc_state->old_change));
	}

	binaryheap_free(fc_state->heap);
	pfree(fc_state);
}

/*
 * 清理事务的内容，通常在事务提交或中止后。
 */
static void fc_ReorderBufferCleanupTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	bool		fc_found;
	dlist_mutable_iter fc_iter;

	/* 清理子事务及其更改 */
	dlist_foreach_modify(fc_iter, &fc_txn->subtxns)
	{
		ReorderBufferTXN *fc_subtxn;

		fc_subtxn = dlist_container(ReorderBufferTXN, node, fc_iter.cur);

		/*
		 * 子事务始终与顶级事务关联，即使它们最初发生在另一个子事务内部，因此我们在这里不会递归超过一个级别。
		 */
		Assert(rbtxn_is_known_subxact(fc_subtxn));
		Assert(fc_subtxn->nsubtxns == 0);

		fc_ReorderBufferCleanupTXN(fc_rb, fc_subtxn);
	}

	/* 清理事务中的更改 */
	dlist_foreach_modify(fc_iter, &fc_txn->changes)
	{
		ReorderBufferChange *fc_change;

		fc_change = dlist_container(ReorderBufferChange, node, fc_iter.cur);

		/* 检查我们没有混合来自不同事务的更改。 */
		Assert(fc_change->txn == fc_txn);

		ReorderBufferReturnChange(fc_rb, fc_change, true);
	}

	/*
	 * 清理我们为解码目录快照访问存储的tuplecids。
	 * 它们总是存储在顶级事务中。
	 */
	dlist_foreach_modify(fc_iter, &fc_txn->tuplecids)
	{
		ReorderBufferChange *fc_change;

		fc_change = dlist_container(ReorderBufferChange, node, fc_iter.cur);

		/* 检查我们没有混合来自不同事务的更改。 */
		Assert(fc_change->txn == fc_txn);
		Assert(fc_change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);

		ReorderBufferReturnChange(fc_rb, fc_change, true);
	}

	/*
	 * 如果设置，清理基础快照。
	 */
	if (fc_txn->base_snapshot != NULL)
	{
		SnapBuildSnapDecRefcount(fc_txn->base_snapshot);
		dlist_delete(&fc_txn->base_snapshot_node);
	}

	/*
	 * 清理最后一次流式运行的快照。
	 */
	if (fc_txn->snapshot_now != NULL)
	{
		Assert(rbtxn_is_streamed(fc_txn));
		fc_ReorderBufferFreeSnap(fc_rb, fc_txn->snapshot_now);
	}

	/*
	 * 从其包含列表中删除事务。
	 *
	 * 注意：如果事务被称为subxact，我们将从其父级的已知子xact列表中删除事务；这会使父级的nsubxacts计数过高，但我们不在意。否则，我们正在从按LSN顺序的顶级事务列表中删除事务。
	 */
	dlist_delete(&fc_txn->node);

	/* 现在从缓冲区中移除引用 */
	hash_search(fc_rb->by_txn,
				(void *) &fc_txn->xid,
				HASH_REMOVE,
				&fc_found);
	Assert(fc_found);

	/* 删除溢出到磁盘的条目 */
	if (rbtxn_is_serialized(fc_txn))
		fc_ReorderBufferRestoreCleanup(fc_rb, fc_txn);

	/* 释放 */
	fc_ReorderBufferReturnTXN(fc_rb, fc_txn);
}

/*
 * 丢弃事务（和子事务）的变化，无论是在流式传输后还是在 PREPARE 时解码它们。保留其余信息 -
 * 事务，tuplecids，无效性和快照。
 *
 * 我们在准备时间解码事务后额外移除 tuplecids，因为我们只需要在回滚或提交准备时执行无效化。
 *
 * 'txn_prepared' 表示我们在准备时间解码了事务。
 */
static void fc_ReorderBufferTruncateTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn, bool fc_txn_prepared)
{
	dlist_mutable_iter fc_iter;

	/* 清理子事务及其更改 */
	dlist_foreach_modify(fc_iter, &fc_txn->subtxns)
	{
		ReorderBufferTXN *fc_subtxn;

		fc_subtxn = dlist_container(ReorderBufferTXN, node, fc_iter.cur);

		/*
		 * 子事务始终与顶级事务关联，即使它们最初发生在另一个子事务内部，因此我们在这里不会递归超过一个级别。
		 */
		Assert(rbtxn_is_known_subxact(fc_subtxn));
		Assert(fc_subtxn->nsubtxns == 0);

		fc_ReorderBufferTruncateTXN(fc_rb, fc_subtxn, fc_txn_prepared);
	}

	/* 清理事务中的更改 */
	dlist_foreach_modify(fc_iter, &fc_txn->changes)
	{
		ReorderBufferChange *fc_change;

		fc_change = dlist_container(ReorderBufferChange, node, fc_iter.cur);

		/* 检查我们没有混合来自不同事务的更改。 */
		Assert(fc_change->txn == fc_txn);

		/* 从其包含列表中移除变化 */
		dlist_delete(&fc_change->node);

		ReorderBufferReturnChange(fc_rb, fc_change, true);
	}

	/*
	 * 将事务标记为已流式传输。
	 *
	 * 顶级事务（标识为 (toptxn==NULL)）始终被标记为流式传输，即使它不包含任何变化（也就是说，当
	 * 所有变化都在子事务中时）。
	 *
	 * 对于子事务，只有在其中有变化时我们才会将其标记为流式传输。
	 *
	 * 我们这样做是因为回滚 - 我们不想发送下游未意识到的 XID 的回滚。当然，它总是知道
	 * 顶级 xact（我们在所有消息中发送 XID），但我们从不流式传输空子xact 的 XID。
	 */
	if ((!fc_txn_prepared) && ((!fc_txn->toptxn) || (fc_txn->nentries_mem != 0)))
		fc_txn->txn_flags |= RBTXN_IS_STREAMED;

	if (fc_txn_prepared)
	{
		/*
		 * 如果这是一个准备好的事务，清理我们为解码目录快照访问存储的 tuplecids。它们始终存储在
		 * 顶级事务中。
		 */
		dlist_foreach_modify(fc_iter, &fc_txn->tuplecids)
		{
			ReorderBufferChange *fc_change;

			fc_change = dlist_container(ReorderBufferChange, node, fc_iter.cur);

			/* 检查我们没有混合来自不同事务的更改。 */
			Assert(fc_change->txn == fc_txn);
			Assert(fc_change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);

			/* 从其包含列表中移除变化。 */
			dlist_delete(&fc_change->node);

			ReorderBufferReturnChange(fc_rb, fc_change, true);
		}
	}

	/*
	 * 销毁 (relfilenode, ctid) 哈希表，以便我们不会泄漏任何
	 * 内存。我们也可以保持哈希表并用新的 ctid 值更新它，
	 * 但这似乎更加简单并且就目前而言足够好。
	 */
	if (fc_txn->tuplecid_hash != NULL)
	{
		hash_destroy(fc_txn->tuplecid_hash);
		fc_txn->tuplecid_hash = NULL;
	}

	/* 如果此事务被序列化，则清理磁盘空间。 */
	if (rbtxn_is_serialized(fc_txn))
	{
		fc_ReorderBufferRestoreCleanup(fc_rb, fc_txn);
		fc_txn->txn_flags &= ~RBTXN_IS_SERIALIZED;

		/*
		 * 我们设置这个标志以指示事务是否曾经被序列化。
		 * 我们需要这个来准确更新统计信息，否则同一
		 * 事务可能会被多次计数为序列化。
		 */
		fc_txn->txn_flags |= RBTXN_IS_SERIALIZED_CLEAR;
	}

	/* 还要重置事务中的条目数量 */
	fc_txn->nentries_mem = 0;
	fc_txn->nentries = 0;
}

/*
 * 构建一个 (relfilenode, ctid) -> (cmin, cmax) 映射的哈希，用于
 * HeapTupleSatisfiesHistoricMVCC。
 */
static void fc_ReorderBufferBuildTupleCidHash(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	dlist_iter	fc_iter;
	HASHCTL		fc_hash_ctl;

	if (!rbtxn_has_catalog_changes(fc_txn) || dlist_is_empty(&fc_txn->tuplecids))
		return;

	fc_hash_ctl.keysize = sizeof(ReorderBufferTupleCidKey);
	fc_hash_ctl.entrysize = sizeof(ReorderBufferTupleCidEnt);
	fc_hash_ctl.hcxt = fc_rb->context;

	/*
	 * 创建一个具有确切数量的要存储的 tuplecids 的哈希，从
	 * 开始
	 */
	fc_txn->tuplecid_hash =
		hash_create("ReorderBufferTupleCid", fc_txn->ntuplecids, &fc_hash_ctl,
					HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);

	dlist_foreach(fc_iter, &fc_txn->tuplecids)
	{
		ReorderBufferTupleCidKey fc_key;
		ReorderBufferTupleCidEnt *fc_ent;
		bool		fc_found;
		ReorderBufferChange *fc_change;

		fc_change = dlist_container(ReorderBufferChange, node, fc_iter.cur);

		Assert(fc_change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);

		/* 注意填充问题 */
		memset(&fc_key, 0, sizeof(ReorderBufferTupleCidKey));

		fc_key.relnode = fc_change->data.tuplecid.node;

		ItemPointerCopy(&fc_change->data.tuplecid.tid,
						&fc_key.tid);

		fc_ent = (ReorderBufferTupleCidEnt *)
			hash_search(fc_txn->tuplecid_hash,
						(void *) &fc_key,
						HASH_ENTER,
						&fc_found);
		if (!fc_found)
		{
			fc_ent->cmin = fc_change->data.tuplecid.cmin;
			fc_ent->cmax = fc_change->data.tuplecid.cmax;
			fc_ent->combocid = fc_change->data.tuplecid.combocid;
		}
		else
		{
			/*
			 * 也许我们在这个事务中已经看到了这个元组，但
			 * 如果是这样，它必须具有相同的 cmin。
			 */
			Assert(fc_ent->cmin == fc_change->data.tuplecid.cmin);

			/*
			 * cmax 可能最初无效，但一旦设置，它只能增长，
			 * 且绝不会再次变为无效。
			 */
			Assert((fc_ent->cmax == InvalidCommandId) ||
				   ((fc_change->data.tuplecid.cmax != InvalidCommandId) &&
					(fc_change->data.tuplecid.cmax > fc_ent->cmax)));
			fc_ent->cmax = fc_change->data.tuplecid.cmax;
		}
	}
}

/*
 * 复制提供的快照，以便我们可以私下修改。这样做是为了
 * 使得修改目录的事务可以查看中间目录状态。
 */
static Snapshot fc_ReorderBufferCopySnap(ReorderBuffer *fc_rb, Snapshot fc_orig_snap,
					  ReorderBufferTXN *fc_txn, CommandId fc_cid)
{
	Snapshot	fc_snap;
	dlist_iter	fc_iter;
	int			fc_i = 0;
	Size		fc_size;

	fc_size = sizeof(SnapshotData) +
		sizeof(TransactionId) * fc_orig_snap->xcnt +
		sizeof(TransactionId) * (fc_txn->nsubtxns + 1);

	fc_snap = MemoryContextAllocZero(fc_rb->context, fc_size);
	memcpy(fc_snap, fc_orig_snap, sizeof(SnapshotData));

	fc_snap->copied = true;
	fc_snap->active_count = 1;		/* 标记为活动状态，以免被释放 */
	fc_snap->regd_count = 0;
	fc_snap->xip = (TransactionId *) (fc_snap + 1);

	memcpy(fc_snap->xip, fc_orig_snap->xip, sizeof(TransactionId) * fc_snap->xcnt);

	/*
	 * snap->subxip 包含所有属于我们事务的 txid，我们
	 * 需要通过 cmin/cmax 检查它们。这就是我们将顶层
	 * 事务存储在其中的原因。
	 */
	fc_snap->subxip = fc_snap->xip + fc_snap->xcnt;
	fc_snap->subxip[fc_i++] = fc_txn->xid;

	/*
	 * 当子事务中止时，subxcnt 不会减少，因此需要手动计数。
	 * 由于这是一个上界，因此可以安全地用于上述分配。
	 */
	fc_snap->subxcnt = 1;

	dlist_foreach(fc_iter, &fc_txn->subtxns)
	{
		ReorderBufferTXN *fc_sub_txn;

		fc_sub_txn = dlist_container(ReorderBufferTXN, node, fc_iter.cur);
		fc_snap->subxip[fc_i++] = fc_sub_txn->xid;
		fc_snap->subxcnt++;
	}

	/* 排序，以便我们可以稍后进行 bsearch() */
	qsort(fc_snap->subxip, fc_snap->subxcnt, sizeof(TransactionId), xidComparator);

	/* 存储指定的当前 CommandId */
	fc_snap->curcid = fc_cid;

	return fc_snap;
}

/*
 * 释放之前 ReorderBufferCopySnap 的快照
 */
static void fc_ReorderBufferFreeSnap(ReorderBuffer *fc_rb, Snapshot fc_snap)
{
	if (fc_snap->copied)
		pfree(fc_snap);
	else
		SnapBuildSnapDecRefcount(fc_snap);
}

/*
 * 如果事务是（部分）流式传输的，我们需要以“流式”
 * 方式准备或提交它。也就是说，我们首先流传输事务的剩余部分，
 * 然后根据情况调用 stream_prepare 或 stream_commit 消息。
 */
static void fc_ReorderBufferStreamCommit(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	/* 我们应该只对之前流式传输的事务调用此函数 */
	Assert(rbtxn_is_streamed(fc_txn));

	fc_ReorderBufferStreamTXN(fc_rb, fc_txn);

	if (rbtxn_prepared(fc_txn))
	{
		/*
		 * 注意，即使检测到并发中止，我们也会发送流准备。
		 * 有关更多信息，请参见 DecodePrepare。
		 */
		fc_rb->stream_prepare(fc_rb, fc_txn, fc_txn->final_lsn);

		/*
		 * 这是一个已准备的事务，是两阶段提交的一部分。
		 * 完整清理将在 COMMIT PREPAREDs 过程中进行，因此现在
		 * 只需通过去除更改和 tuple_cids 来截断事务。
		 */
		fc_ReorderBufferTruncateTXN(fc_rb, fc_txn, true);
		/* 重置 CheckXidAlive */
		CheckXidAlive = InvalidTransactionId;
	}
	else
	{
		fc_rb->stream_commit(fc_rb, fc_txn, fc_txn->final_lsn);
		fc_ReorderBufferCleanupTXN(fc_rb, fc_txn);
	}
}

/*
 * 设置xid以检测并发中止。
 *
 * 在流式传输进行中的事务或解码准备好的事务时，(子)事务可能会被并发中止。在这种情况下，如果(子)事务有目录更新，那么我们可能会使用错误的目录版本解码元组。例如，假设有一个目录元组为(xmin: 500, xmax: 0)。现在，事务501更新了目录元组，之后我们将有两个元组(xmin: 500, xmax: 501)和(xmin: 501, xmax: 0)。现在，如果501被中止并且另一个事务如502更新了同一目录元组，那么第一个元组将被更改为(xmin: 500, xmax: 502)。所以，问题在于当我们尝试解码在501中插入/更新的元组时，我们将看到目录元组(xmin: 500, xmax: 502)是可见的，因为它将认为该元组是被xid 502删除的，而这对我们的快照不可见。当我们尝试与该目录元组解码时，可能会导致错误结果或崩溃。因此，有必要检测并发中止，以允许进行中事务的流式传输或准备事务的解码。
 *
 * 为了检测并发中止，我们将CheckXidAlive设置为当前(子)事务的xid，该更改属于此事务。在目录扫描期间，我们可以检查xid的状态，如果它被中止，我们将报告特定错误，以便我们可以停止当前事务的流式传输，并在此类错误上丢弃已经流式传输的更改。我们可能已经流式传输了一些被中止的(子)事务的更改，但这没关系，因为当我们解码中止时，我们将流式传输中止消息以截断订阅者中的更改。类似地，对于准备好的事务，如果检测到并发中止，我们将停止解码，然后在遇到回滚准备时回滚更改。请参见DecodePrepare。
 */
static inline void fc_SetupCheckXidLive(TransactionId fc_xid)
{
	/*
	 * 如果输入事务id已经设置为CheckXidAlive，那么什么都不需要做。
	 */
	if (TransactionIdEquals(CheckXidAlive, fc_xid))
		return;

	/*
	 * 如果尚未提交，则设置CheckXidAlive。我们不检查xid是否被中止。这将在目录访问期间发生。
	 */
	if (!TransactionIdDidCommit(fc_xid))
		CheckXidAlive = fc_xid;
	else
		CheckXidAlive = InvalidTransactionId;
}

/*
 * ReorderBufferProcessTXN的帮助函数，用于应用更改。
 */
static inline void fc_ReorderBufferApplyChange(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
						 Relation fc_relation, ReorderBufferChange *fc_change,
						 bool fc_streaming)
{
	if (fc_streaming)
		fc_rb->stream_change(fc_rb, fc_txn, fc_relation, fc_change);
	else
		fc_rb->apply_change(fc_rb, fc_txn, fc_relation, fc_change);
}

/*
 * ReorderBufferProcessTXN的帮助函数，用于应用截断。
 */
static inline void fc_ReorderBufferApplyTruncate(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
						   int fc_nrelations, Relation *fc_relations,
						   ReorderBufferChange *fc_change, bool fc_streaming)
{
	if (fc_streaming)
		fc_rb->stream_truncate(fc_rb, fc_txn, fc_nrelations, fc_relations, fc_change);
	else
		fc_rb->apply_truncate(fc_rb, fc_txn, fc_nrelations, fc_relations, fc_change);
}

/*
 * ReorderBufferProcessTXN的帮助函数，用于应用消息。
 */
static inline void fc_ReorderBufferApplyMessage(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
						  ReorderBufferChange *fc_change, bool fc_streaming)
{
	if (fc_streaming)
		fc_rb->stream_message(fc_rb, fc_txn, fc_change->lsn, true,
						   fc_change->data.msg.prefix,
						   fc_change->data.msg.message_size,
						   fc_change->data.msg.message);
	else
		fc_rb->message(fc_rb, fc_txn, fc_change->lsn, true,
					fc_change->data.msg.prefix,
					fc_change->data.msg.message_size,
					fc_change->data.msg.message);
}

/*
 * 函数用于在当前流的末尾存储命令id和快照，以便在发送下一个流时可以重用相同的内容。
 */
static inline void fc_ReorderBufferSaveTXNSnapshot(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
							 Snapshot fc_snapshot_now, CommandId fc_command_id)
{
	fc_txn->command_id = fc_command_id;

	/* 如果已经被复制，则避免复制。 */
	if (fc_snapshot_now->copied)
		fc_txn->snapshot_now = fc_snapshot_now;
	else
		fc_txn->snapshot_now = fc_ReorderBufferCopySnap(fc_rb, fc_snapshot_now,
												  fc_txn, fc_command_id);
}

/*
 * ReorderBufferProcessTXN 的辅助函数，用于处理流式事务的并发中止。
 * 这将重置 TXN，以便可以用于流式处理正在处理的事务的剩余数据。
 * 当子事务中止时，这可能会发生，而我们仍然希望继续处理主事务或其他子事务的数据。
 */
static void fc_ReorderBufferResetTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
					  Snapshot fc_snapshot_now,
					  CommandId fc_command_id,
					  XLogRecPtr fc_last_lsn,
					  ReorderBufferChange *fc_specinsert)
{
	/* 丢弃我们刚刚流过的更改 */
	fc_ReorderBufferTruncateTXN(fc_rb, fc_txn, rbtxn_prepared(fc_txn));

	/* 释放为 toast 重建分配的所有资源 */
	fc_ReorderBufferToastReset(fc_rb, fc_txn);

	/* 如果不为 NULL，则返回规范插入更改 */
	if (fc_specinsert != NULL)
	{
		ReorderBufferReturnChange(fc_rb, fc_specinsert, true);
		fc_specinsert = NULL;
	}

	/*
	 * 对于流式情况，停止流并记住命令 ID 和
	 * 流式运行的快照。
	 */
	if (rbtxn_is_streamed(fc_txn))
	{
		fc_rb->stream_stop(fc_rb, fc_txn, fc_last_lsn);
		fc_ReorderBufferSaveTXNSnapshot(fc_rb, fc_txn, fc_snapshot_now, fc_command_id);
	}
}

/*
 * ReorderBufferReplay 和 ReorderBufferStreamTXN 的辅助函数。
 *
 * 将事务（及其子事务）的数据发送到
 * 输出插件。我们遍历主事务和子事务（使用 k 路
 * 合并）并按 lsn 顺序重放更改。
 *
 * 如果 streaming 为 true，则将使用流 API 发送数据。
 *
 * 注意：某些参数上的“volatile”标记是为了避免在
 * 函数内部使用 PG_TRY 时出现问题。
 */
static void fc_ReorderBufferProcessTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
						XLogRecPtr fc_commit_lsn,
						volatile Snapshot fc_snapshot_now,
						volatile CommandId fc_command_id,
						bool fc_streaming)
{
	bool		fc_using_subtxn;
	MemoryContext fc_ccxt = CurrentMemoryContext;
	ReorderBufferIterTXNState *volatile fc_iterstate = NULL;
	volatile XLogRecPtr fc_prev_lsn = InvalidXLogRecPtr;
	ReorderBufferChange *volatile fc_specinsert = NULL;
	volatile bool fc_stream_started = false;
	ReorderBufferTXN *volatile fc_curtxn = NULL;

	/* 构建数据以便能够查找目录元组的 CommandIds */
	fc_ReorderBufferBuildTupleCidHash(fc_rb, fc_txn);

	/* 设置初始快照 */
	SetupHistoricSnapshot(fc_snapshot_now, fc_txn->tuplecid_hash);

	/*
	 * 解码需要访问系统缓存等，而这些又使用
	 * 重型锁等。因此我们需要有足够的状态来
	 * 跟踪这些。最简单的方法是简单地在内部使用事务。
	 * 这也使我们能够轻松地强制检查没有写入
	 * 数据库的迹象，通过检查 xid 分配。
	 *
	 * 当通过 SQL SRF 调用时，已经启动了一个事务，
	 * 所以在那里启动一个显式的子事务。
	 */
	fc_using_subtxn = IsTransactionOrTransactionBlock();

	PG_TRY();
	{
		ReorderBufferChange *fc_change;

		if (fc_using_subtxn)
			BeginInternalSubTransaction(fc_streaming ? "stream" : "replay");
		else
			StartTransactionCommand();

		/*
		 * 我们只需要为非流式
		 * 事务发送 begin/begin-prepare。
		 */
		if (!fc_streaming)
		{
			if (rbtxn_prepared(fc_txn))
				fc_rb->begin_prepare(fc_rb, fc_txn);
			else
				fc_rb->begin(fc_rb, fc_txn);
		}

		fc_ReorderBufferIterTXNInit(fc_rb, fc_txn, &fc_iterstate);
		while ((fc_change = fc_ReorderBufferIterTXNNext(fc_rb, fc_iterstate)) != NULL)
		{
			Relation	fc_relation = NULL;
			Oid			fc_reloid;

			CHECK_FOR_INTERRUPTS();

			/*
			 * 在处理第一个更改之前，我们不能调用启动流回调。
			 */
			if (fc_prev_lsn == InvalidXLogRecPtr)
			{
				if (fc_streaming)
				{
					fc_txn->origin_id = fc_change->origin_id;
					fc_rb->stream_start(fc_rb, fc_txn, fc_change->lsn);
					fc_stream_started = true;
				}
			}

			/*
			 * 强制从多个子事务合并的更改的正确顺序。
			 * 由于 MULTI_INSERT xlog 记录，更改可能具有相同的 LSN。
			 */
			Assert(fc_prev_lsn == InvalidXLogRecPtr || fc_prev_lsn <= fc_change->lsn);

			fc_prev_lsn = fc_change->lsn;

			/*
			 * 设置当前 xid 以检测并发中止。这在我们解码
			 * 更改而 COMMIT 记录尚未处理的情况下是必需的。
			 */
			if (fc_streaming || rbtxn_prepared(fc_change->txn))
			{
				fc_curtxn = fc_change->txn;
				fc_SetupCheckXidLive(fc_curtxn->xid);
			}

			switch (fc_change->action)
			{
				case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:

					/*
					 * 投机性插入的确认已到达。简单地
					 * 将其用作正常记录。它将在 INSERT
					 * 处理结束时进行清理。
					 */
					if (fc_specinsert == NULL)
						elog(ERROR, "invalid ordering of speculative insertion changes");
					Assert(fc_specinsert->data.tp.oldtuple == NULL);
					fc_change = fc_specinsert;
					fc_change->action = REORDER_BUFFER_CHANGE_INSERT;

					/* intentionally fall through */
				case REORDER_BUFFER_CHANGE_INSERT:
				case REORDER_BUFFER_CHANGE_UPDATE:
				case REORDER_BUFFER_CHANGE_DELETE:
					Assert(fc_snapshot_now);

					fc_reloid = RelidByRelfilenode(fc_change->data.tp.relnode.spcNode,
												fc_change->data.tp.relnode.relNode);

					/*
					 * 没有数据的映射目录元组，在目录表正在被重写时发出。我们
					 * 可能无法查找relfilenode，因为relmapper没有“历史”
					 * 视图，这与解码期间的正常目录形成对比。因此，重复的重写
					 * 可能会导致查找失败。这没关系，因为我们反正并不解码
					 * 目录变更。通常这些元组在下面会被跳过，但我们无法识别
					 * 表是否应该被逻辑记录，而不将relfilenode映射到oid。
					 */
					if (fc_reloid == InvalidOid &&
						fc_change->data.tp.newtuple == NULL &&
						fc_change->data.tp.oldtuple == NULL)
						goto change_done;
					else if (fc_reloid == InvalidOid)
						elog(ERROR, "could not map filenode \"%s\" to relation OID",
							 relpathperm(fc_change->data.tp.relnode,
										 MAIN_FORKNUM));

					fc_relation = RelationIdGetRelation(fc_reloid);

					if (!RelationIsValid(fc_relation))
						elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
							 fc_reloid,
							 relpathperm(fc_change->data.tp.relnode,
										 MAIN_FORKNUM));

					if (!RelationIsLogicallyLogged(fc_relation))
						goto change_done;

					/*
					 * 除非插件请求，否则忽略在DDL过程中创建的临时堆。
					 */
					if (fc_relation->rd_rel->relrewrite && !fc_rb->output_rewrites)
						goto change_done;

					/*
					 * 目前完全忽略序列变更。大多数情况下，它们不使用
					 * 我们理解的记录来记录变更，因此处理少数几个
					 * 情况没有意义。
					 */
					if (fc_relation->rd_rel->relkind == RELKIND_SEQUENCE)
						goto change_done;

					/* 用户触发的变更 */
					if (!IsToastRelation(fc_relation))
					{
						fc_ReorderBufferToastReplace(fc_rb, fc_txn, fc_relation, fc_change);
						fc_ReorderBufferApplyChange(fc_rb, fc_txn, fc_relation, fc_change,
												 fc_streaming);

						/*
						 * 只有在确定已经不再需要时，才清除重新组装的
						 * toast块。元组的创建者告诉我们。
						 */
						if (fc_change->data.tp.clear_toast_afterwards)
							fc_ReorderBufferToastReset(fc_rb, fc_txn);
					}
					/* 我们对toast删除不感兴趣 */
					else if (fc_change->action == REORDER_BUFFER_CHANGE_INSERT)
					{
						/*
						 * 需要在内存中重新组装完整的toasted Datum，以确保
						 * 块在我们完成之前不会被重复使用，从
						 * 此事务的变更列表中移除它。否则它将在从
						 * 磁盘恢复缓存数据时被释放/重复使用。
						 */
						Assert(fc_change->data.tp.newtuple != NULL);

						dlist_delete(&fc_change->node);
						fc_ReorderBufferToastAppendChunk(fc_rb, fc_txn, fc_relation,
													  fc_change);
					}

			change_done:

					/*
					 * 如果推测插入被确认，记录就不再需要了。
					 */
					if (fc_specinsert != NULL)
					{
						ReorderBufferReturnChange(fc_rb, fc_specinsert, true);
						fc_specinsert = NULL;
					}

					if (RelationIsValid(fc_relation))
					{
						RelationClose(fc_relation);
						fc_relation = NULL;
					}
					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:

					/*
					 * 推测插入通过延迟处理插入直到确认记录到达来处理。
					 * 为此，我们只是将记录从链中解除链接，以便在从
					 * 磁盘恢复缓存数据时不会被释放/重复使用。
					 *
					 * 在面对并发目录变更时这是安全的，因为相关关系不能在
					 * 推测插入与确认之间发生变化，这得益于CheckTableNotInUse()
					 * 和锁定机制。
					 */

					/* 清除一个待处理（因此失败） 的推测 */
					if (fc_specinsert != NULL)
					{
						ReorderBufferReturnChange(fc_rb, fc_specinsert, true);
						fc_specinsert = NULL;
					}

					/* 并记住待处理的插入 */
					dlist_delete(&fc_change->node);
					fc_specinsert = fc_change;
					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:

					/*
					 * 推测插入到达，进行中止。因此清理
					 * specinsert元组和toast哈希。
					 *
					 * 请注意，我们为每个toast条目获得推测中止变更，
					 * 但我们只需要在第一次处理主表时进行清理。
					 */
					if (fc_specinsert != NULL)
					{
						/*
						 * 在处理一个全新的元组之前，必须清理toast哈希，以避免
						 * 对之前元组的toast块产生混淆。
						 */
						Assert(fc_change->data.tp.clear_toast_afterwards);
						fc_ReorderBufferToastReset(fc_rb, fc_txn);

						/* 我们不再需要这个记录了。 */
						ReorderBufferReturnChange(fc_rb, fc_specinsert, true);
						fc_specinsert = NULL;
					}
					break;

				case REORDER_BUFFER_CHANGE_TRUNCATE:
					{
						int			fc_i;
						int			fc_nrelids = fc_change->data.truncate.nrelids;
						int			fc_nrelations = 0;
						Relation   *fc_relations;

						fc_relations = palloc0(fc_nrelids * sizeof(Relation));
						for (fc_i = 0; fc_i < fc_nrelids; fc_i++)
						{
							Oid			fc_relid = fc_change->data.truncate.relids[fc_i];
							Relation	fc_relation;

							fc_relation = RelationIdGetRelation(fc_relid);

							if (!RelationIsValid(fc_relation))
								elog(ERROR, "could not open relation with OID %u", fc_relid);

							if (!RelationIsLogicallyLogged(fc_relation))
								continue;

							fc_relations[fc_nrelations++] = fc_relation;
						}

						/* 应用截断。 */
						fc_ReorderBufferApplyTruncate(fc_rb, fc_txn, fc_nrelations,
												   fc_relations, fc_change,
												   fc_streaming);

						for (fc_i = 0; fc_i < fc_nrelations; fc_i++)
							RelationClose(fc_relations[fc_i]);

						break;
					}

				case REORDER_BUFFER_CHANGE_MESSAGE:
					fc_ReorderBufferApplyMessage(fc_rb, fc_txn, fc_change, fc_streaming);
					break;

				case REORDER_BUFFER_CHANGE_INVALIDATION:
					/* 在本地执行无效化消息 */
					fc_ReorderBufferExecuteInvalidations(fc_change->data.inval.ninvalidations,
													  fc_change->data.inval.invalidations);
					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
					/* 摆脱旧的 */
					TeardownHistoricSnapshot(false);

					if (fc_snapshot_now->copied)
					{
						fc_ReorderBufferFreeSnap(fc_rb, fc_snapshot_now);
						fc_snapshot_now =
							fc_ReorderBufferCopySnap(fc_rb, fc_change->data.snapshot,
												  fc_txn, fc_command_id);
					}

					/*
					 * 从磁盘恢复，需要小心不要重复释放。
					 * 我们可以为此引入引用计数，但目前看来这种情况不够频繁，不用太在意。
					 */
					else if (fc_change->data.snapshot->copied)
					{
						fc_snapshot_now =
							fc_ReorderBufferCopySnap(fc_rb, fc_change->data.snapshot,
												  fc_txn, fc_command_id);
					}
					else
					{
						fc_snapshot_now = fc_change->data.snapshot;
					}

					/* 并继续使用新的 */
					SetupHistoricSnapshot(fc_snapshot_now, fc_txn->tuplecid_hash);
					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
					Assert(fc_change->data.command_id != InvalidCommandId);

					if (fc_command_id < fc_change->data.command_id)
					{
						fc_command_id = fc_change->data.command_id;

						if (!fc_snapshot_now->copied)
						{
							/* 我们不再使用全局的 */
							fc_snapshot_now = fc_ReorderBufferCopySnap(fc_rb, fc_snapshot_now,
																 fc_txn, fc_command_id);
						}

						fc_snapshot_now->curcid = fc_command_id;

						TeardownHistoricSnapshot(false);
						SetupHistoricSnapshot(fc_snapshot_now, fc_txn->tuplecid_hash);
					}

					break;

				case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
					elog(ERROR, "tuplecid value in changequeue");
					break;
			}
		}

		/* 投机性插入记录现在必须被释放 */
		Assert(!fc_specinsert);

		/* 清理迭代器 */
		fc_ReorderBufferIterTXNFinish(fc_rb, fc_iterstate);
		fc_iterstate = NULL;

		/*
		 * 更新总事务计数和事务及其子事务处理的总字节数。确保不会多次计入流式事务。
		 *
		 * 注意，统计计算必须在ReorderBufferIterTXNFinish之后进行，因为它释放了已在ReorderBufferIterTXNNext中计入的序列化更改。
		 */
		if (!rbtxn_is_streamed(fc_txn))
			fc_rb->totalTxns++;

		fc_rb->totalBytes += fc_txn->total_size;

		/*
		 * 当前更改已完成，视流式模式发送此组更改的最后消息。
		 */
		if (fc_streaming)
		{
			if (fc_stream_started)
			{
				fc_rb->stream_stop(fc_rb, fc_txn, fc_prev_lsn);
				fc_stream_started = false;
			}
		}
		else
		{
			/*
			 * 调用PREPARE（对于两阶段事务）或COMMIT（对于常规事务）。
			 */
			if (rbtxn_prepared(fc_txn))
				fc_rb->prepare(fc_rb, fc_txn, fc_commit_lsn);
			else
				fc_rb->commit(fc_rb, fc_txn, fc_commit_lsn);
		}

		/* 这只是对坏输出插件行为的合理性检查 */
		if (GetCurrentTransactionIdIfAny() != InvalidTransactionId)
			elog(ERROR, "output plugin used XID %u",
				 GetCurrentTransactionId());

		/*
		 * 记住命令ID和快照，以便在流式模式下进行下一组更改。
		 */
		if (fc_streaming)
			fc_ReorderBufferSaveTXNSnapshot(fc_rb, fc_txn, fc_snapshot_now, fc_command_id);
		else if (fc_snapshot_now->copied)
			fc_ReorderBufferFreeSnap(fc_rb, fc_snapshot_now);

		/* 清理 */
		TeardownHistoricSnapshot(false);

		/*
		 * 以整体方式中止当前（子）事务具有正确的语义。我们希望在这里获取的所有锁都被释放，而不是重新分配给父事务，并且我们不希望任何数据库访问产生持久影响。
		 */
		AbortCurrentTransaction();

		/* 确保没有缓存污染 */
		fc_ReorderBufferExecuteInvalidations(fc_txn->ninvalidations, fc_txn->invalidations);

		if (fc_using_subtxn)
			RollbackAndReleaseCurrentSubTransaction();

		/*
		 * 我们在这里是出于四个原因之一：1. 解码一个进行中的事务。2. 解码一个准备好的事务。3. 解码一个（部分）流式准备好的事务。4. 解码一个已提交的事务。
		 *
		 * 对于1，我们允许通过删除已流式传输的更改来截断事务数据，但仍保留其他内容，如失效、快照和元组CID。对于2和3，我们指示ReorderBufferTruncateTXN进行更复杂的事务数据截断，因为除了提交外，整个事务已经被解码。对于4，由于整个事务已经被解码，我们可以完全清理事务重排缓冲区。
		 */
		if (fc_streaming || rbtxn_prepared(fc_txn))
		{
			fc_ReorderBufferTruncateTXN(fc_rb, fc_txn, rbtxn_prepared(fc_txn));
			/* 重置 CheckXidAlive */
			CheckXidAlive = InvalidTransactionId;
		}
		else
			fc_ReorderBufferCleanupTXN(fc_rb, fc_txn);
	}
	PG_CATCH();
	{
		MemoryContext fc_ecxt = MemoryContextSwitchTo(fc_ccxt);
		ErrorData  *fc_errdata = CopyErrorData();

		/* TODO: 封装PG_TRY和PG_CATCH块中的清理 */
		if (fc_iterstate)
			fc_ReorderBufferIterTXNFinish(fc_rb, fc_iterstate);

		TeardownHistoricSnapshot(true);

		/*
		 * 强制在无效事务外部发生缓存失效，以防止目录访问，因为我们刚捕获到一个错误。
		 */
		AbortCurrentTransaction();

		/* 确保没有缓存污染 */
		fc_ReorderBufferExecuteInvalidations(fc_txn->ninvalidations,
										  fc_txn->invalidations);

		if (fc_using_subtxn)
			RollbackAndReleaseCurrentSubTransaction();

		/*
		 * 错误代码 ERRCODE_TRANSACTION_ROLLBACK 表示我们正在流式传输或准备的 (子) 事务的并发终止。我们需要进行清理并优雅地返回此错误，请参见 SetupCheckXidLive。
		 *
		 * 该错误代码可能由我们在解码过程中调用的回调中的一个抛出，因此我们需要确保仅在我们以流式模式发送数据并且流式传输尚未完成时，或者在我们在两阶段提交过程中发送 PREPARE 数据时优雅地返回。
		 */
		if (fc_errdata->sqlerrcode == ERRCODE_TRANSACTION_ROLLBACK &&
			(fc_stream_started || rbtxn_prepared(fc_txn)))
		{
			/* curtxn 必须为流式或准备事务设置 */
			Assert(fc_curtxn);

			/* 清理临时错误状态。 */
			FlushErrorState();
			FreeErrorData(fc_errdata);
			fc_errdata = NULL;
			fc_curtxn->concurrent_abort = true;

			/* 重置 TXN 以允许流式传输剩余数据。 */
			fc_ReorderBufferResetTXN(fc_rb, fc_txn, fc_snapshot_now,
								  fc_command_id, fc_prev_lsn,
								  fc_specinsert);
		}
		else
		{
			fc_ReorderBufferCleanupTXN(fc_rb, fc_txn);
			MemoryContextSwitchTo(fc_ecxt);
			PG_RE_THROW();
		}
	}
	PG_END_TRY();
}

/*
 * 执行事务及其未中止的子事务的重放。
 *
 * 子事务之前必须由 ReorderBufferCommitChild() 处理，即使之前已通过 ReorderBufferAssignChild 分配给顶级事务。
 *
 * 此接口在读取准备或顶级提交时被调用，无论是流式事务还是非流式事务。
 */
static void fc_ReorderBufferReplay(ReorderBufferTXN *fc_txn,
					ReorderBuffer *fc_rb, TransactionId fc_xid,
					XLogRecPtr fc_commit_lsn, XLogRecPtr fc_end_lsn,
					TimestampTz fc_commit_time,
					RepOriginId fc_origin_id, XLogRecPtr fc_origin_lsn)
{
	Snapshot	fc_snapshot_now;
	CommandId	fc_command_id = FirstCommandId;

	fc_txn->final_lsn = fc_commit_lsn;
	fc_txn->end_lsn = fc_end_lsn;
	fc_txn->xact_time.commit_time = fc_commit_time;
	fc_txn->origin_id = fc_origin_id;
	fc_txn->origin_lsn = fc_origin_lsn;

	/*
	 * 如果事务是（部分）流式传输的，我们需要以“流式”方式提交它。也就是说，我们首先流式传输事务的其余部分，然后调用 stream_commit 消息。
	 *
	 * 在事务中的所有信息（原始 ID、LSN 等）存储完毕后调用，以避免直接传递该信息。
	 */
	if (rbtxn_is_streamed(fc_txn))
	{
		fc_ReorderBufferStreamCommit(fc_rb, fc_txn);
		return;
	}

	/*
	 * 如果此事务没有快照，它未对数据库做出任何更改，因此没有什么可解码的。请注意，如果存在任何子事务，ReorderBufferCommitChild 将转移任何快照。
	 */
	if (fc_txn->base_snapshot == NULL)
	{
		Assert(fc_txn->ninvalidations == 0);

		/*
		 * 在提交之前删除此事务可能会导致重新启动 LSN 的计算不正确。请参见 SnapBuildProcessRunningXacts。
		 */
		if (!rbtxn_prepared(fc_txn))
			fc_ReorderBufferCleanupTXN(fc_rb, fc_txn);
		return;
	}

	fc_snapshot_now = fc_txn->base_snapshot;

	/* 处理并将更改发送到输出插件。 */
	fc_ReorderBufferProcessTXN(fc_rb, fc_txn, fc_commit_lsn, fc_snapshot_now,
							fc_command_id, false);
}

/*
 * 提交事务。
 *
 * 请参阅 ReorderBufferReplay() 的注释。
 */
void ReorderBufferCommit(ReorderBuffer *fc_rb, TransactionId fc_xid,
					XLogRecPtr fc_commit_lsn, XLogRecPtr fc_end_lsn,
					TimestampTz fc_commit_time,
					RepOriginId fc_origin_id, XLogRecPtr fc_origin_lsn)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, InvalidXLogRecPtr,
								false);

	/* 未知事务，没什么可重放的 */
	if (fc_txn == NULL)
		return;

	fc_ReorderBufferReplay(fc_txn, fc_rb, fc_xid, fc_commit_lsn, fc_end_lsn, fc_commit_time,
						fc_origin_id, fc_origin_lsn);
}

/*
 * 记录事务的准备信息。
 */
bool ReorderBufferRememberPrepareInfo(ReorderBuffer *fc_rb, TransactionId fc_xid,
								 XLogRecPtr fc_prepare_lsn, XLogRecPtr fc_end_lsn,
								 TimestampTz fc_prepare_time,
								 RepOriginId fc_origin_id, XLogRecPtr fc_origin_lsn)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, InvalidXLogRecPtr, false);

	/* 未知事务，没什么可做的 */
	if (fc_txn == NULL)
		return false;

	/*
	 * 记住准备信息，以便稍后在我们跳过准备时由 commit prepared 使用。
	 */
	fc_txn->final_lsn = fc_prepare_lsn;
	fc_txn->end_lsn = fc_end_lsn;
	fc_txn->xact_time.prepare_time = fc_prepare_time;
	fc_txn->origin_id = fc_origin_id;
	fc_txn->origin_lsn = fc_origin_lsn;

	return true;
}

/* 记住我们已跳过准备 */
void ReorderBufferSkipPrepare(ReorderBuffer *fc_rb, TransactionId fc_xid)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, InvalidXLogRecPtr, false);

	/* 未知事务，没什么可做的 */
	if (fc_txn == NULL)
		return;

	fc_txn->txn_flags |= RBTXN_SKIPPED_PREPARE;
}

/*
 * 准备一个两阶段事务。
 *
 * 请参阅 ReorderBufferReplay() 的注释。
 */
void ReorderBufferPrepare(ReorderBuffer *fc_rb, TransactionId fc_xid,
					 char *fc_gid)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, InvalidXLogRecPtr,
								false);

	/* 未知事务，没什么可重放的 */
	if (fc_txn == NULL)
		return;

	fc_txn->txn_flags |= RBTXN_PREPARE;
	fc_txn->gid = pstrdup(fc_gid);

	/* 现在准备信息必须已在事务中更新。 */
	Assert(fc_txn->final_lsn != InvalidXLogRecPtr);

	fc_ReorderBufferReplay(fc_txn, fc_rb, fc_xid, fc_txn->final_lsn, fc_txn->end_lsn,
						fc_txn->xact_time.prepare_time, fc_txn->origin_id, fc_txn->origin_lsn);

	/*
	 * 我们为并发中止的事务发送准备，以便在解码并发送 rollback prepared 时，下游应该能够回滚此事务。请参见 DecodePrepare 顶部的注释。
	 *
	 * 注意，对于并发中止 + 流式传输的情况，已经在上述 ReorderBufferReplay 调用中发送了 stream_prepare。
	 */
	if (fc_txn->concurrent_abort && !rbtxn_is_streamed(fc_txn))
		fc_rb->prepare(fc_rb, fc_txn, fc_txn->final_lsn);
}

/*
 * 这用于处理 COMMIT/ROLLBACK PREPARED。
 */
void ReorderBufferFinishPrepared(ReorderBuffer *fc_rb, TransactionId fc_xid,
							XLogRecPtr fc_commit_lsn, XLogRecPtr fc_end_lsn,
							XLogRecPtr fc_two_phase_at,
							TimestampTz fc_commit_time, RepOriginId fc_origin_id,
							XLogRecPtr fc_origin_lsn, char *fc_gid, bool fc_is_commit)
{
	ReorderBufferTXN *fc_txn;
	XLogRecPtr	fc_prepare_end_lsn;
	TimestampTz fc_prepare_time;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, fc_commit_lsn, false);

	/* 未知事务，没什么可做的 */
	if (fc_txn == NULL)
		return;

	/*
	 * 到此时，txn 已具有准备记录信息，请记住以便稍后用于回滚。
	 */
	fc_prepare_end_lsn = fc_txn->end_lsn;
	fc_prepare_time = fc_txn->xact_time.prepare_time;

	/* 在 txn 中添加 gid */
	fc_txn->gid = pstrdup(fc_gid);

	/*
	 * 可能在准备时未解码此事务，原因可能是到那时我们没有一致的快照，
	 * 或两个阶段没有启用，或者它之前已解码但我们已重启。
	 * 只有在之前未解码时我们才需要发送准备。如果尚未完成，
	 * 我们不需要为中止解码 xact。
	 */
	if ((fc_txn->final_lsn < fc_two_phase_at) && fc_is_commit)
	{
		fc_txn->txn_flags |= RBTXN_PREPARE;

		/*
		 * 即使我们跳过准备，准备信息也必须在 txn 中更新。
		 */
		Assert(fc_txn->final_lsn != InvalidXLogRecPtr);

		/*
		 * 到此时，txn 已具有准备记录信息，使用此信息非常重要，
		 * 以便下游获得准确的信息。如果相反，我们在这里传递了提交信息，
		 * 那么下游可能会表现得好像它已经在重启后重新播放了提交准备。
		 */
		fc_ReorderBufferReplay(fc_txn, fc_rb, fc_xid, fc_txn->final_lsn, fc_txn->end_lsn,
							fc_txn->xact_time.prepare_time, fc_txn->origin_id, fc_txn->origin_lsn);
	}

	fc_txn->final_lsn = fc_commit_lsn;
	fc_txn->end_lsn = fc_end_lsn;
	fc_txn->xact_time.commit_time = fc_commit_time;
	fc_txn->origin_id = fc_origin_id;
	fc_txn->origin_lsn = fc_origin_lsn;

	if (fc_is_commit)
		fc_rb->commit_prepared(fc_rb, fc_txn, fc_commit_lsn);
	else
		fc_rb->rollback_prepared(fc_rb, fc_txn, fc_prepare_end_lsn, fc_prepare_time);

	/* 清理：确保没有缓存污染 */
	fc_ReorderBufferExecuteInvalidations(fc_txn->ninvalidations,
									  fc_txn->invalidations);
	fc_ReorderBufferCleanupTXN(fc_rb, fc_txn);
}

/*
 * 中止一个可能已进行更改的事务。需要先对子事务调用，
 * 然后对顶层 xid 调用。
 *
 * 注意：在这里处理的事务必须是主动中止的（即已生成中止记录）。
 * 隐式中止的事务通过 ReorderBufferAbortOld() 处理；我们
 * 不感兴趣但已提交的事务通过 ReorderBufferForget() 处理。
 *
 * 此函数清除此事务及其内容的内存和磁盘。
 */
void ReorderBufferAbort(ReorderBuffer *fc_rb, TransactionId fc_xid, XLogRecPtr fc_lsn)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, InvalidXLogRecPtr,
								false);

	/* 未知，无需移除 */
	if (fc_txn == NULL)
		return;

	/* 对于流式事务，通知远程节点有关中止的情况。 */
	if (rbtxn_is_streamed(fc_txn))
	{
		fc_rb->stream_abort(fc_rb, fc_txn, fc_lsn);

		/*
		 * 我们可能已经解码了此事务的更改，这可能根据当前事务的视图
		 * 加载缓存（考虑此事务中发生的 DDL）。我们不希望未来事务
		 * 的解码使用这些缓存条目，因此执行无效化。
		 */
		if (fc_txn->ninvalidations > 0)
			ReorderBufferImmediateInvalidation(fc_rb, fc_txn->ninvalidations,
											   fc_txn->invalidations);
	}

	/* 仅外观调整... */
	fc_txn->final_lsn = fc_lsn;

	/* 移除潜在的磁盘数据，并解除分配 */
	fc_ReorderBufferCleanupTXN(fc_rb, fc_txn);
}

/*
 * 中止所有实际上不再运行的事务，因为服务器重启了。
 *
 * 注意：这些确实必须是由于服务器崩溃/立即重启而中止的事务，
 * 因为我们在这里不处理无效化。
 */
void ReorderBufferAbortOld(ReorderBuffer *fc_rb, TransactionId fc_oldestRunningXid)
{
	dlist_mutable_iter fc_it;

	/*
	 * 遍历所有（潜在的）顶层 TXNs，中止所有比可能正在运行的
	 * 更旧的事务。一旦找到第一个仍然存活的事务，我们就停止，
	 * 可能还有一些事务在之前获取了 xid，但开始写入较晚，
	 * 但这种情况不太可能，它们将在对该函数的后续调用中被清除。
	 */
	dlist_foreach_modify(fc_it, &fc_rb->toplevel_by_lsn)
	{
		ReorderBufferTXN *fc_txn;

		fc_txn = dlist_container(ReorderBufferTXN, node, fc_it.cur);

		if (TransactionIdPrecedes(fc_txn->xid, fc_oldestRunningXid))
		{
			elog(DEBUG2, "aborting old transaction %u", fc_txn->xid);

			/* 通知远程节点有关崩溃/立即重启的情况。 */
			if (rbtxn_is_streamed(fc_txn))
				fc_rb->stream_abort(fc_rb, fc_txn, InvalidXLogRecPtr);

			/* 移除潜在的磁盘数据，并解除分配此 tx */
			fc_ReorderBufferCleanupTXN(fc_rb, fc_txn);
		}
		else
			return;
	}
}

/*
 * 如果我们不关心交易的内容，则遗忘交易的内容。需要先对子事务调用，然后对顶层 xid 调用。
 *
 * 这与 ReorderBufferAbort() 有显著不同，因为已提交的事务需要与中止的事务有所区别，因为它们可能已经修改了目录。
 *
 * 请注意，这只能在刚读取到事务提交时调用，而不能更早；否则，后续记录引用此 xid 可能会不完全重新创建该事务。
 */
void ReorderBufferForget(ReorderBuffer *fc_rb, TransactionId fc_xid, XLogRecPtr fc_lsn)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, InvalidXLogRecPtr,
								false);

	/* 未知，无需遗忘 */
	if (fc_txn == NULL)
		return;

	/* 对于流式事务，通知远程节点有关中止的情况。 */
	if (rbtxn_is_streamed(fc_txn))
		fc_rb->stream_abort(fc_rb, fc_txn, fc_lsn);

	/* 仅外观调整... */
	fc_txn->final_lsn = fc_lsn;

	/*
	 * 如果有缓存失效消息，请处理。即使我们对事务的内容不感兴趣，它仍然可能操作了目录，我们需要根据其更新缓存。
	 */
	if (fc_txn->base_snapshot != NULL && fc_txn->ninvalidations > 0)
		ReorderBufferImmediateInvalidation(fc_rb, fc_txn->ninvalidations,
										   fc_txn->invalidations);
	else
		Assert(fc_txn->ninvalidations == 0);

	/* 移除潜在的磁盘数据，并解除分配 */
	fc_ReorderBufferCleanupTXN(fc_rb, fc_txn);
}

/*
 * 对于那些需要跳过的事务使缓存失效，以防目录被事务操作。
 *
 * 请注意，这是一个针对预处理事务的特殊目的函数，在我们决定跳过时，我们不想清理 TXN。详见 DecodePrepare。
 */
void ReorderBufferInvalidate(ReorderBuffer *fc_rb, TransactionId fc_xid, XLogRecPtr fc_lsn)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, InvalidXLogRecPtr,
								false);

	/* 未知，无需做任何事情 */
	if (fc_txn == NULL)
		return;

	/*
	 * 如果有缓存失效消息，请处理。即使我们对事务的内容不感兴趣，它仍然可能操作了目录，我们需要根据其更新缓存。
	 */
	if (fc_txn->base_snapshot != NULL && fc_txn->ninvalidations > 0)
		ReorderBufferImmediateInvalidation(fc_rb, fc_txn->ninvalidations,
										   fc_txn->invalidations);
	else
		Assert(fc_txn->ninvalidations == 0);
}


/*
 * 执行在解码事务上下文之外发生的失效。这通常发生在无 xid 提交（参见 RecordTransactionCommit()）或在不感兴趣的事务中的失效（通过 ReorderBufferForget()）。
 */
void ReorderBufferImmediateInvalidation(ReorderBuffer *fc_rb, uint32 fc_ninvalidations,
								   SharedInvalidationMessage *fc_invalidations)
{
	bool		fc_use_subtxn = IsTransactionOrTransactionBlock();
	int			fc_i;

	if (fc_use_subtxn)
		BeginInternalSubTransaction("replay");

	/*
	 * 强制使失效在有效事务之外发生 - 这样条目将仅被标记为无效，而无需访问目录。这是有利的，因为我们无需设置访问目录所需的完整状态。
	 */
	if (fc_use_subtxn)
		AbortCurrentTransaction();

	for (fc_i = 0; fc_i < fc_ninvalidations; fc_i++)
		LocalExecuteInvalidationMessage(&fc_invalidations[fc_i]);

	if (fc_use_subtxn)
		RollbackAndReleaseCurrentSubTransaction();
}

/*
 * 告诉 reorderbuffer 在 WAL 流中看到的一个 xid。每个 XLogRecord->xl_xid 中的 xid 至少必须调用一次（记录中的其他地方可以，但不一定必须通过此处）。
 *
 * Reorderbuffer 保持有关事务的某些数据结构，按 LSN 顺序排列，以提高效率。为此，它必须知道事务在 WAL 中首次被看到的时间。由于许多类型的记录在逻辑解码中并不实际有意义，因此它们不一定会通过此处。
 */
void ReorderBufferProcessXid(ReorderBuffer *fc_rb, TransactionId fc_xid, XLogRecPtr fc_lsn)
{
	/* 许多记录将不会分配 xid，在这里集中检查 */
	if (fc_xid != InvalidTransactionId)
		fc_ReorderBufferTXNByXid(fc_rb, fc_xid, true, NULL, fc_lsn, true);
}

/*
 * 向该事务添加一个新的快照，该快照只能在 lsn 'lsn' 之后使用，因为先前的快照无法正确描述后续行的目录。
 */
void ReorderBufferAddSnapshot(ReorderBuffer *fc_rb, TransactionId fc_xid,
						 XLogRecPtr fc_lsn, Snapshot fc_snap)
{
	ReorderBufferChange *fc_change = ReorderBufferGetChange(fc_rb);

	fc_change->data.snapshot = fc_snap;
	fc_change->action = REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT;

	ReorderBufferQueueChange(fc_rb, fc_xid, fc_lsn, fc_change, false);
}

/*
 * 设置事务的基础快照。
 *
 * 如果我们知道 xid 是一个子事务，则在顶级事务上设置基础快照。
 */
void ReorderBufferSetBaseSnapshot(ReorderBuffer *fc_rb, TransactionId fc_xid,
							 XLogRecPtr fc_lsn, Snapshot fc_snap)
{
	ReorderBufferTXN *fc_txn;
	bool		fc_is_new;

	AssertArg(fc_snap != NULL);

	/*
	 * 获取要操作的事务。如果我们知道它是一个子事务，
	 * 则在其顶级事务上操作。
	 */
	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, true, &fc_is_new, fc_lsn, true);
	if (rbtxn_is_known_subxact(fc_txn))
		fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_txn->toplevel_xid, false,
									NULL, InvalidXLogRecPtr, false);
	Assert(fc_txn->base_snapshot == NULL);

	fc_txn->base_snapshot = fc_snap;
	fc_txn->base_snapshot_lsn = fc_lsn;
	dlist_push_tail(&fc_rb->txns_by_base_snapshot_lsn, &fc_txn->base_snapshot_node);

	fc_AssertTXNLsnOrder(fc_rb);
}

/*
 * 在这个 CommandId 的变化流中访问目录。
 *
 * 只能对命令 id > 1 调用
 */
void ReorderBufferAddNewCommandId(ReorderBuffer *fc_rb, TransactionId fc_xid,
							 XLogRecPtr fc_lsn, CommandId fc_cid)
{
	ReorderBufferChange *fc_change = ReorderBufferGetChange(fc_rb);

	fc_change->data.command_id = fc_cid;
	fc_change->action = REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID;

	ReorderBufferQueueChange(fc_rb, fc_xid, fc_lsn, fc_change, false);
}

/*
 * 更新内存计数器以考虑新变更或移除的变更。
 *
 * 我们更新两个计数器 - 在重排缓冲区中，以及在包含变更的
 * 事务中。重排缓冲区计数器允许我们快速决定是否达到了内存限制，
 * 事务计数器允许我们快速选择最大的事务进行驱逐。
 *
 * 当启用流式处理时，我们需要更新顶级事务计数器 - 我们实际上
 * 不关心子事务，因为我们无法单独流式处理它们，并且我们
 * 只选择顶级事务进行驱逐。因此，只有顶级事务才重要。
 */
static void fc_ReorderBufferChangeMemoryUpdate(ReorderBuffer *fc_rb,
								ReorderBufferChange *fc_change,
								bool fc_addition, Size fc_sz)
{
	ReorderBufferTXN *fc_txn;
	ReorderBufferTXN *fc_toptxn;

	Assert(fc_change->txn);

	/*
	 * 忽略元组 CID 更改，因为在达到内存限制时这些不会被驱逐。
	 * 所以我们只是不计算它们，因为这可能轻易引发一次无意义的
	 * 溢出尝试。
	 */
	if (fc_change->action == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID)
		return;

	fc_txn = fc_change->txn;

	/*
	 * 也更新顶级的总大小。这稍后用于计算解码统计。
	 */
	if (fc_txn->toptxn != NULL)
		fc_toptxn = fc_txn->toptxn;
	else
		fc_toptxn = fc_txn;

	if (fc_addition)
	{
		fc_txn->size += fc_sz;
		fc_rb->size += fc_sz;

		/* 更新顶级事务中的总大小。 */
		fc_toptxn->total_size += fc_sz;
	}
	else
	{
		Assert((fc_rb->size >= fc_sz) && (fc_txn->size >= fc_sz));
		fc_txn->size -= fc_sz;
		fc_rb->size -= fc_sz;

		/* 更新顶级事务中的总大小。 */
		fc_toptxn->total_size -= fc_sz;
	}

	Assert(fc_txn->size <= fc_rb->size);
}

/*
 * 添加新的 (relfilenode, tid) -> (cmin, cmax) 映射。
 *
 * 我们不将此变更类型包含在内存计算中，因为我们
 * 将 CIDs 保存在一个单独的列表中，并且在达到内存限制时
 * 不驱逐它们。
 */
void ReorderBufferAddNewTupleCids(ReorderBuffer *fc_rb, TransactionId fc_xid,
							 XLogRecPtr fc_lsn, RelFileNode fc_node,
							 ItemPointerData fc_tid, CommandId fc_cmin,
							 CommandId fc_cmax, CommandId fc_combocid)
{
	ReorderBufferChange *fc_change = ReorderBufferGetChange(fc_rb);
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, true, NULL, fc_lsn, true);

	fc_change->data.tuplecid.node = fc_node;
	fc_change->data.tuplecid.tid = fc_tid;
	fc_change->data.tuplecid.cmin = fc_cmin;
	fc_change->data.tuplecid.cmax = fc_cmax;
	fc_change->data.tuplecid.combocid = fc_combocid;
	fc_change->lsn = fc_lsn;
	fc_change->txn = fc_txn;
	fc_change->action = REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID;

	dlist_push_tail(&fc_txn->tuplecids, &fc_change->node);
	fc_txn->ntuplecids++;
}

/*
 * 累积无效通知以便稍后执行。
 *
 * 这需要为每个 XLOG_XACT_INVALIDATIONS 消息调用，并
 * 在顶级事务中累积所有无效通知消息（如果可用），
 * 否则在当前事务中，以及以重排缓冲区中的变更形式。
 * 我们需要以变更的形式记录它，以便只执行所需的无效
 * 通知，而不是在每次 CommandId 增加时执行所有的无效
 * 通知。我们还需要在 txn 缓冲区中累积这些，因为在某些
 * 情况下我们跳过处理事务（见 ReorderBufferForget），
 * 我们需要一起执行所有的无效通知。
 */
void ReorderBufferAddInvalidations(ReorderBuffer *fc_rb, TransactionId fc_xid,
							  XLogRecPtr fc_lsn, Size fc_nmsgs,
							  SharedInvalidationMessage *fc_msgs)
{
	ReorderBufferTXN *fc_txn;
	MemoryContext fc_oldcontext;
	ReorderBufferChange *fc_change;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, true, NULL, fc_lsn, true);

	fc_oldcontext = MemoryContextSwitchTo(fc_rb->context);

	/*
	 * 如果可用，收集所有顶层事务下的无效通知，
	 * 以便我们可以一起执行它们。见此函数顶部的
	 * 注释。
	 */
	if (fc_txn->toptxn)
		fc_txn = fc_txn->toptxn;

	Assert(fc_nmsgs > 0);

	/* 累积无效通知。 */
	if (fc_txn->ninvalidations == 0)
	{
		fc_txn->ninvalidations = fc_nmsgs;
		fc_txn->invalidations = (SharedInvalidationMessage *)
			palloc(sizeof(SharedInvalidationMessage) * fc_nmsgs);
		memcpy(fc_txn->invalidations, fc_msgs,
			   sizeof(SharedInvalidationMessage) * fc_nmsgs);
	}
	else
	{
		fc_txn->invalidations = (SharedInvalidationMessage *)
			repalloc(fc_txn->invalidations, sizeof(SharedInvalidationMessage) *
					 (fc_txn->ninvalidations + fc_nmsgs));

		memcpy(fc_txn->invalidations + fc_txn->ninvalidations, fc_msgs,
			   fc_nmsgs * sizeof(SharedInvalidationMessage));
		fc_txn->ninvalidations += fc_nmsgs;
	}

	fc_change = ReorderBufferGetChange(fc_rb);
	fc_change->action = REORDER_BUFFER_CHANGE_INVALIDATION;
	fc_change->data.inval.ninvalidations = fc_nmsgs;
	fc_change->data.inval.invalidations = (SharedInvalidationMessage *)
		palloc(sizeof(SharedInvalidationMessage) * fc_nmsgs);
	memcpy(fc_change->data.inval.invalidations, fc_msgs,
		   sizeof(SharedInvalidationMessage) * fc_nmsgs);

	ReorderBufferQueueChange(fc_rb, fc_xid, fc_lsn, fc_change, false);

	MemoryContextSwitchTo(fc_oldcontext);
}

/*
 * 应用我们知道的所有无效通知。在变化流的
 * 这一点上，我们可能只需要部分通知，但我们不知道
 * 这些是哪部分。
 */
static void fc_ReorderBufferExecuteInvalidations(uint32 fc_nmsgs, SharedInvalidationMessage *fc_msgs)
{
	int			fc_i;

	for (fc_i = 0; fc_i < fc_nmsgs; fc_i++)
		LocalExecuteInvalidationMessage(&fc_msgs[fc_i]);
}

/*
 * 将事务标记为包含目录更改
 */
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *fc_rb, TransactionId fc_xid,
								  XLogRecPtr fc_lsn)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, true, NULL, fc_lsn, true);

	fc_txn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;

	/*
	 * 如果其一个子事务有变更，将顶级事务标记为也有目录变更，以便
	 * ReorderBufferBuildTupleCidHash可以方便地仅检查顶级事务并决定
	 * 是否构建哈希表。
	 */
	if (fc_txn->toptxn != NULL)
		fc_txn->toptxn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
}

/*
 * 查询一个事务是否已经*已知*包含目录变更。在提交之前这可能是错误的！
 */
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *fc_rb, TransactionId fc_xid)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false, NULL, InvalidXLogRecPtr,
								false);
	if (fc_txn == NULL)
		return false;

	return rbtxn_has_catalog_changes(fc_txn);
}

/*
 * ReorderBufferXidHasBaseSnapshot
 *		我们是否已经为给定的事务/子事务设置了基快照？
 */
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *fc_rb, TransactionId fc_xid)
{
	ReorderBufferTXN *fc_txn;

	fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_xid, false,
								NULL, InvalidXLogRecPtr, false);

	/* 事务尚未被识别，因此没有快照 */
	if (fc_txn == NULL)
		return false;

	/* 已知的子事务？改为操作顶级事务 */
	if (rbtxn_is_known_subxact(fc_txn))
		fc_txn = fc_ReorderBufferTXNByXid(fc_rb, fc_txn->toplevel_xid, false,
									NULL, InvalidXLogRecPtr, false);

	return fc_txn->base_snapshot != NULL;
}


/*
 * ---------------------------------------
 * 磁盘序列化支持
 * ---------------------------------------
 */

/*
 * 确保IO缓冲区大于或等于sz。
 */
static void fc_ReorderBufferSerializeReserve(ReorderBuffer *fc_rb, Size fc_sz)
{
	if (!fc_rb->outbufsize)
	{
		fc_rb->outbuf = MemoryContextAlloc(fc_rb->context, fc_sz);
		fc_rb->outbufsize = fc_sz;
	}
	else if (fc_rb->outbufsize < fc_sz)
	{
		fc_rb->outbuf = repalloc(fc_rb->outbuf, fc_sz);
		fc_rb->outbufsize = fc_sz;
	}
}

/*
 * 找到最大的事务（顶级或子事务）以驱逐（溢出到磁盘）。
 *
 * XXX 如果有许多子事务，这可能会非常缓慢，因为我们必须遍历所有
 * 子事务。我们可以通过以下方式来改善： (a) 维护一些按变更数量
 * 排序的事务的辅助结构, (b) 不寻找完全最大的事务，而是例如
 * 寻找占用至少部分内存限制的事务， (c) 一次驱逐多个事务，例如
 * 释放特定部分的内存限制（例如 50%）。
 */
static ReorderBufferTXN * fc_ReorderBufferLargestTXN(ReorderBuffer *fc_rb)
{
	HASH_SEQ_STATUS fc_hash_seq;
	ReorderBufferTXNByIdEnt *fc_ent;
	ReorderBufferTXN *fc_largest = NULL;

	hash_seq_init(&fc_hash_seq, fc_rb->by_txn);
	while ((fc_ent = hash_seq_search(&fc_hash_seq)) != NULL)
	{
		ReorderBufferTXN *fc_txn = fc_ent->txn;

		/* 如果当前事务更大，记住它 */
		if ((!fc_largest) || (fc_txn->size > fc_largest->size))
			fc_largest = fc_txn;
	}

	Assert(fc_largest);
	Assert(fc_largest->size > 0);
	Assert(fc_largest->size <= fc_rb->size);

	return fc_largest;
}

/*
 * 找到最大的顶级事务以驱逐（通过流）。
 *
 * 这可以看作是ReorderBufferLargestTXN的优化版本，它应该给我们
 * 相同的事务（因为我们不更新子事务的内存账户，所以它总是0）。
 * 但是我们可以简单地遍历有限数量的具有基快照的顶级事务。
 * 选择没有基快照的事务没有意义，因为我们不解码此类事务。
 *
 * 请注意，我们跳过包含不完整更改的事务。这里有优化的空间，
 * 使我们可以选择具有不完整更改的最大事务。但这会使代码和设计
 * 变得相当复杂，可能不值得。如果我们计划 потоковую 处理包含不完整变更的
 * 事务，那么我们需要找到一种方法来部分流/截断内存中的事务更改
 * 并构建一种机制以部分截断溢出的文件。
 * 此外，无论何时我们部分流事务，我们都需要维护上一个流的lsn，
 * 下次需要从该段和WAL中的偏移量恢复。当我们从顶级事务流
 * 变更并按子事务恢复时，我们甚至需要记住我们上一次流的
 * 变更的子事务。
 */
static ReorderBufferTXN * fc_ReorderBufferLargestTopTXN(ReorderBuffer *fc_rb)
{
	dlist_iter	fc_iter;
	Size		fc_largest_size = 0;
	ReorderBufferTXN *fc_largest = NULL;

	/* 找到具有基快照的最大顶级事务。 */
	dlist_foreach(fc_iter, &fc_rb->txns_by_base_snapshot_lsn)
	{
		ReorderBufferTXN *fc_txn;

		fc_txn = dlist_container(ReorderBufferTXN, base_snapshot_node, fc_iter.cur);

		/* 不能是子事务 */
		Assert(!rbtxn_is_known_subxact(fc_txn));
		/* base_snapshot 必须被设置 */
		Assert(fc_txn->base_snapshot != NULL);

		if ((fc_largest == NULL || fc_txn->total_size > fc_largest_size) &&
			(fc_txn->total_size > 0) && !(rbtxn_has_partial_change(fc_txn)))
		{
			fc_largest = fc_txn;
			fc_largest_size = fc_txn->total_size;
		}
	}

	return fc_largest;
}

/*
 * 检查是否达到了 logical_decoding_work_mem 限制，如果是，选择最大的
 * (子)事务进行驱逐，并将其更改溢出到磁盘，直到我们低于内存限制。
 *
 * XXX 在这一点上，我们选择事务直到我们低于内存限制，但我们也可能
 * 采用更复杂的驱逐策略——例如驱逐足够的事务以释放某个比例（例如 50%）
 * 的内存限制。
 */
static void fc_ReorderBufferCheckMemoryLimit(ReorderBuffer *fc_rb)
{
	ReorderBufferTXN *fc_txn;

	/* 如果我们没有超过内存限制则退出 */
	if (fc_rb->size < logical_decoding_work_mem * 1024L)
		return;

	/*
	 * 循环直到我们低于内存限制。人们可能会认为只需驱逐最大的
	 * (子)事务就能低于内存限制，因为假设选择的事务至少与最近的更改
	 * 一样大（这导致我们超过内存限制）。然而，这并不成立，因为用户可以在
	 * 最近的更改之前将 logical_decoding_work_mem 减小到更小的值。
	 */
	while (fc_rb->size >= logical_decoding_work_mem * 1024L)
	{
		/*
		 * 选择最大的事务并通过流式传输将其从内存驱逐，如果可能的话。
		 * 否则，溢出到磁盘。
		 */
		if (fc_ReorderBufferCanStartStreaming(fc_rb) &&
			(fc_txn = fc_ReorderBufferLargestTopTXN(fc_rb)) != NULL)
		{
			/* 我们知道必须有一个，因为大小不为零 */
			Assert(fc_txn && !fc_txn->toptxn);
			Assert(fc_txn->total_size > 0);
			Assert(fc_rb->size >= fc_txn->total_size);

			fc_ReorderBufferStreamTXN(fc_rb, fc_txn);
		}
		else
		{
			/*
			 * 选择最大的事务（或子事务）并通过将其序列化到磁盘将其从内存驱逐。
			 */
			fc_txn = fc_ReorderBufferLargestTXN(fc_rb);

			/* 我们知道必须有一个，因为大小不为零 */
			Assert(fc_txn);
			Assert(fc_txn->size > 0);
			Assert(fc_rb->size >= fc_txn->size);

			fc_ReorderBufferSerializeTXN(fc_rb, fc_txn);
		}

		/*
		 * 驱逐后，事务应该在内存中没有任何条目，并且更改应使用 0 字节。
		 */
		Assert(fc_txn->size == 0);
		Assert(fc_txn->nentries_mem == 0);
	}

	/* 我们现在必须低于内存限制。 */
	Assert(fc_rb->size < logical_decoding_work_mem * 1024L);
}

/*
 * 将大事务（及其子事务）数据溢出到磁盘。
 */
static void fc_ReorderBufferSerializeTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	dlist_iter	fc_subtxn_i;
	dlist_mutable_iter fc_change_i;
	int			fc_fd = -1;
	XLogSegNo	fc_curOpenSegNo = 0;
	Size		fc_spilled = 0;
	Size		fc_size = fc_txn->size;

	elog(DEBUG2, "spill %u changes in XID %u to disk",
		 (uint32) fc_txn->nentries_mem, fc_txn->xid);

	/* 对所有子 TX 也做相同的操作 */
	dlist_foreach(fc_subtxn_i, &fc_txn->subtxns)
	{
		ReorderBufferTXN *fc_subtxn;

		fc_subtxn = dlist_container(ReorderBufferTXN, node, fc_subtxn_i.cur);
		fc_ReorderBufferSerializeTXN(fc_rb, fc_subtxn);
	}

	/* 序列化变更流 */
	dlist_foreach_modify(fc_change_i, &fc_txn->changes)
	{
		ReorderBufferChange *fc_change;

		fc_change = dlist_container(ReorderBufferChange, node, fc_change_i.cur);

		/*
		 * 根据起始 lsn 存储到所属段，不要跨多个段分割
		 */
		if (fc_fd == -1 ||
			!XLByteInSeg(fc_change->lsn, fc_curOpenSegNo, wal_segment_size))
		{
			char		fc_path[MAXPGPATH];

			if (fc_fd != -1)
				CloseTransientFile(fc_fd);

			XLByteToSeg(fc_change->lsn, fc_curOpenSegNo, wal_segment_size);

			/*
			 * 这里不需要关心 TLI，只在单次运行中使用，因此每个 LSN
			 * 只映射到特定的 WAL 记录。
			 */
			fc_ReorderBufferSerializedPath(fc_path, MyReplicationSlot, fc_txn->xid,
										fc_curOpenSegNo);

			/* 打开段，如有必要则创建它 */
			fc_fd = OpenTransientFile(fc_path,
								   O_CREAT | O_WRONLY | O_APPEND | PG_BINARY);

			if (fc_fd < 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not open file \"%s\": %m", fc_path)));
		}

		fc_ReorderBufferSerializeChange(fc_rb, fc_txn, fc_fd, fc_change);
		dlist_delete(&fc_change->node);
		ReorderBufferReturnChange(fc_rb, fc_change, true);

		fc_spilled++;
	}

	/* 仅在我们溢出任何内容时更新统计信息 */
	if (fc_spilled)
	{
		fc_rb->spillCount += 1;
		fc_rb->spillBytes += fc_size;

		/* 不考虑已经序列化的事务 */
		fc_rb->spillTxns += (rbtxn_is_serialized(fc_txn) || rbtxn_is_serialized_clear(fc_txn)) ? 0 : 1;

		/* 更新解码统计信息 */
		UpdateDecodingStats((LogicalDecodingContext *) fc_rb->private_data);
	}

	Assert(fc_spilled == fc_txn->nentries_mem);
	Assert(dlist_is_empty(&fc_txn->changes));
	fc_txn->nentries_mem = 0;
	fc_txn->txn_flags |= RBTXN_IS_SERIALIZED;

	if (fc_fd != -1)
		CloseTransientFile(fc_fd);
}

/*
 * 将单个更改序列化到磁盘。
 */
static void fc_ReorderBufferSerializeChange(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
							 int fc_fd, ReorderBufferChange *fc_change)
{
	ReorderBufferDiskChange *fc_ondisk;
	Size		fc_sz = sizeof(ReorderBufferDiskChange);

	fc_ReorderBufferSerializeReserve(fc_rb, fc_sz);

	fc_ondisk = (ReorderBufferDiskChange *) fc_rb->outbuf;
	memcpy(&fc_ondisk->change, fc_change, sizeof(ReorderBufferChange));

	switch (fc_change->action)
	{
			/* 跳过这些，它们都足够相似 */
		case REORDER_BUFFER_CHANGE_INSERT:
		case REORDER_BUFFER_CHANGE_UPDATE:
		case REORDER_BUFFER_CHANGE_DELETE:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
			{
				char	   *fc_data;
				ReorderBufferTupleBuf *fc_oldtup,
						   *fc_newtup;
				Size		fc_oldlen = 0;
				Size		fc_newlen = 0;

				fc_oldtup = fc_change->data.tp.oldtuple;
				fc_newtup = fc_change->data.tp.newtuple;

				if (fc_oldtup)
				{
					fc_sz += sizeof(HeapTupleData);
					fc_oldlen = fc_oldtup->tuple.t_len;
					fc_sz += fc_oldlen;
				}

				if (fc_newtup)
				{
					fc_sz += sizeof(HeapTupleData);
					fc_newlen = fc_newtup->tuple.t_len;
					fc_sz += fc_newlen;
				}

				/* 确保我们有足够的空间 */
				fc_ReorderBufferSerializeReserve(fc_rb, fc_sz);

				fc_data = ((char *) fc_rb->outbuf) + sizeof(ReorderBufferDiskChange);
				/* 可能在上面重新分配 */
				fc_ondisk = (ReorderBufferDiskChange *) fc_rb->outbuf;

				if (fc_oldlen)
				{
					memcpy(fc_data, &fc_oldtup->tuple, sizeof(HeapTupleData));
					fc_data += sizeof(HeapTupleData);

					memcpy(fc_data, fc_oldtup->tuple.t_data, fc_oldlen);
					fc_data += fc_oldlen;
				}

				if (fc_newlen)
				{
					memcpy(fc_data, &fc_newtup->tuple, sizeof(HeapTupleData));
					fc_data += sizeof(HeapTupleData);

					memcpy(fc_data, fc_newtup->tuple.t_data, fc_newlen);
					fc_data += fc_newlen;
				}
				break;
			}
		case REORDER_BUFFER_CHANGE_MESSAGE:
			{
				char	   *fc_data;
				Size		fc_prefix_size = strlen(fc_change->data.msg.prefix) + 1;

				fc_sz += fc_prefix_size + fc_change->data.msg.message_size +
					sizeof(Size) + sizeof(Size);
				fc_ReorderBufferSerializeReserve(fc_rb, fc_sz);

				fc_data = ((char *) fc_rb->outbuf) + sizeof(ReorderBufferDiskChange);

				/* 可能在上面重新分配 */
				fc_ondisk = (ReorderBufferDiskChange *) fc_rb->outbuf;

				/* 写入包括大小的前缀 */
				memcpy(fc_data, &fc_prefix_size, sizeof(Size));
				fc_data += sizeof(Size);
				memcpy(fc_data, fc_change->data.msg.prefix,
					   fc_prefix_size);
				fc_data += fc_prefix_size;

				/* 写入包括大小的消息 */
				memcpy(fc_data, &fc_change->data.msg.message_size, sizeof(Size));
				fc_data += sizeof(Size);
				memcpy(fc_data, fc_change->data.msg.message,
					   fc_change->data.msg.message_size);
				fc_data += fc_change->data.msg.message_size;

				break;
			}
		case REORDER_BUFFER_CHANGE_INVALIDATION:
			{
				char	   *fc_data;
				Size		fc_inval_size = sizeof(SharedInvalidationMessage) *
				fc_change->data.inval.ninvalidations;

				fc_sz += fc_inval_size;

				fc_ReorderBufferSerializeReserve(fc_rb, fc_sz);
				fc_data = ((char *) fc_rb->outbuf) + sizeof(ReorderBufferDiskChange);

				/* 可能在上面重新分配 */
				fc_ondisk = (ReorderBufferDiskChange *) fc_rb->outbuf;
				memcpy(fc_data, fc_change->data.inval.invalidations, fc_inval_size);
				fc_data += fc_inval_size;

				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
			{
				Snapshot	fc_snap;
				char	   *fc_data;

				fc_snap = fc_change->data.snapshot;

				fc_sz += sizeof(SnapshotData) +
					sizeof(TransactionId) * fc_snap->xcnt +
					sizeof(TransactionId) * fc_snap->subxcnt;

				/* 确保我们有足够的空间 */
				fc_ReorderBufferSerializeReserve(fc_rb, fc_sz);
				fc_data = ((char *) fc_rb->outbuf) + sizeof(ReorderBufferDiskChange);
				/* 可能在上面重新分配 */
				fc_ondisk = (ReorderBufferDiskChange *) fc_rb->outbuf;

				memcpy(fc_data, fc_snap, sizeof(SnapshotData));
				fc_data += sizeof(SnapshotData);

				if (fc_snap->xcnt)
				{
					memcpy(fc_data, fc_snap->xip,
						   sizeof(TransactionId) * fc_snap->xcnt);
					fc_data += sizeof(TransactionId) * fc_snap->xcnt;
				}

				if (fc_snap->subxcnt)
				{
					memcpy(fc_data, fc_snap->subxip,
						   sizeof(TransactionId) * fc_snap->subxcnt);
					fc_data += sizeof(TransactionId) * fc_snap->subxcnt;
				}
				break;
			}
		case REORDER_BUFFER_CHANGE_TRUNCATE:
			{
				Size		fc_size;
				char	   *fc_data;

				/* 考虑被截断关系的 OID */
				fc_size = sizeof(Oid) * fc_change->data.truncate.nrelids;
				fc_sz += fc_size;

				/* 确保我们有足够的空间 */
				fc_ReorderBufferSerializeReserve(fc_rb, fc_sz);

				fc_data = ((char *) fc_rb->outbuf) + sizeof(ReorderBufferDiskChange);
				/* 可能在上面重新分配 */
				fc_ondisk = (ReorderBufferDiskChange *) fc_rb->outbuf;

				memcpy(fc_data, fc_change->data.truncate.relids, fc_size);
				fc_data += fc_size;

				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
			/* ReorderBufferChange 包含所有重要内容 */
			break;
	}

	fc_ondisk->size = fc_sz;

	errno = 0;
	pgstat_report_wait_start(WAIT_EVENT_REORDER_BUFFER_WRITE);
	if (write(fc_fd, fc_rb->outbuf, fc_ondisk->size) != fc_ondisk->size)
	{
		int			fc_save_errno = errno;

		CloseTransientFile(fc_fd);

		/* 如果写入未设置 errno，假设问题是没有磁盘空间 */
		errno = fc_save_errno ? fc_save_errno : ENOSPC;
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not write to data file for XID %u: %m",
						fc_txn->xid)));
	}
	pgstat_report_wait_end();

	/*
	 * 将事务的 final_lsn 随每次我们发送到磁盘的更改保持最新，以便 ReorderBufferRestoreCleanup 正常工作。 （我们过去只在提交和中止记录上执行此操作，但如果系统崩溃导致事务没有其中止记录，这样做就不奏效）。
	 *
	 * 确保不向后移动它。
	 */
	if (fc_txn->final_lsn < fc_change->lsn)
		fc_txn->final_lsn = fc_change->lsn;

	Assert(fc_ondisk->change.action == fc_change->action);
}

/* 如果输出插件支持流式传输则返回 true，否则返回 false。 */
static inline bool fc_ReorderBufferCanStream(ReorderBuffer *fc_rb)
{
	LogicalDecodingContext *fc_ctx = fc_rb->private_data;

	return fc_ctx->streaming;
}

/* 如果流式传输现在可以开始则返回 true，否则返回 false。 */
static inline bool fc_ReorderBufferCanStartStreaming(ReorderBuffer *fc_rb)
{
	LogicalDecodingContext *fc_ctx = fc_rb->private_data;
	SnapBuild  *fc_builder = fc_ctx->snapshot_builder;

	/* 如果未达到一致状态，则我们无法启动流式传输。 */
	if (SnapBuildCurrentState(fc_builder) < SNAPBUILD_CONSISTENT)
		return false;

	/*
	 * 即使启用流式传输，我们也无法立即启动流式传输，因为我们之前解码了该事务，现在只是重新启动。
	 */
	if (fc_ReorderBufferCanStream(fc_rb) &&
		!SnapBuildXactNeedsSkip(fc_builder, fc_ctx->reader->EndRecPtr))
		return true;

	return false;
}

/*
 * 将大事务（及其子事务）的数据发送到输出插件，但使用流式 API。
 */
static void fc_ReorderBufferStreamTXN(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	Snapshot	fc_snapshot_now;
	CommandId	fc_command_id;
	Size		fc_stream_bytes;
	bool		fc_txn_is_streamed;

	/* 子事务永远无法到达这里。 */
	Assert(fc_txn->toptxn == NULL);

	/*
	 * 我们在这里无法对基础快照做出任何假设，类似于 ReorderBufferCommit() 的做法。那依赖于在 ReorderBufferCommitChild() 中从子事务传输 base_snapshot，但在事务进行中时尚未调用该操作。
	 *
	 * 因此，只需遍历子事务并在这里使用相同的逻辑。但我们只需要在事务第一次流式传输时这样做。 之后，我们需要重用上一次运行的快照。
	 *
	 * 与 DecodeCommit 通过 SnapBuildCommittedTxn 将所有子事务的 xids 添加到快照的 xip 数组不同，我们在这里无法做到这一点，但通过 ReorderBufferCopySnap 将它们添加到 subxip 数组中。 这使得至今解码的子事务中所做的目录更改可见。
	 */
	if (fc_txn->snapshot_now == NULL)
	{
		dlist_iter	fc_subxact_i;

		/* 确保该事务是首次流式传输 */
		Assert(!rbtxn_is_streamed(fc_txn));

		/* 开始时我们应该有无效的命令 ID */
		Assert(fc_txn->command_id == InvalidCommandId);

		dlist_foreach(fc_subxact_i, &fc_txn->subtxns)
		{
			ReorderBufferTXN *fc_subtxn;

			fc_subtxn = dlist_container(ReorderBufferTXN, node, fc_subxact_i.cur);
			fc_ReorderBufferTransferSnapToParent(fc_txn, fc_subtxn);
		}

		/*
		 * 如果该事务没有快照，则到现在为止，它对数据库没有任何更改，因此没有任何内容可以解码。
		 */
		if (fc_txn->base_snapshot == NULL)
		{
			Assert(fc_txn->ninvalidations == 0);
			return;
		}

		fc_command_id = FirstCommandId;
		fc_snapshot_now = fc_ReorderBufferCopySnap(fc_rb, fc_txn->base_snapshot,
											 fc_txn, fc_command_id);
	}
	else
	{
		/* 该事务必须已经被流式传输 */
		Assert(rbtxn_is_streamed(fc_txn));

		/*
		 * 不，我们已经从上一次流式传输运行中获得了快照。我们假设新的子事务不能向后移动 LSN，因此无法在之前的分支中击败 LSN 条件（因此无需再次遍历子事务）。 实际上，我们不应该这样做，因为我们可能在子事务进行到一半时使用快照。
		 */
		fc_command_id = fc_txn->command_id;

		/*
		 * 我们无法直接使用 txn->snapshot_now，因为在上一次流式传输运行后，我们可能获得了一些新的子事务。因此，我们需要将它们添加到快照中。
		 */
		fc_snapshot_now = fc_ReorderBufferCopySnap(fc_rb, fc_txn->snapshot_now,
											 fc_txn, fc_command_id);

		/* 释放之前复制的快照。 */
		Assert(fc_txn->snapshot_now->copied);
		fc_ReorderBufferFreeSnap(fc_rb, fc_txn->snapshot_now);
		fc_txn->snapshot_now = NULL;
	}

	/*
	 * 记住这些信息，以便稍后用于更新统计信息。我们无法在这里更新统计信息，因为在处理更改时出现错误会导致统计信息的累计，即使我们尚未流式传输所有更改。
	 */
	fc_txn_is_streamed = rbtxn_is_streamed(fc_txn);
	fc_stream_bytes = fc_txn->total_size;

	/* 处理并将更改发送到输出插件。 */
	fc_ReorderBufferProcessTXN(fc_rb, fc_txn, InvalidXLogRecPtr, fc_snapshot_now,
							fc_command_id, true);

	fc_rb->streamCount += 1;
	fc_rb->streamBytes += fc_stream_bytes;

	/* 不考虑已经流式传输的事务。 */
	fc_rb->streamTxns += (fc_txn_is_streamed) ? 0 : 1;

	/* 更新解码统计信息 */
	UpdateDecodingStats((LogicalDecodingContext *) fc_rb->private_data);

	Assert(dlist_is_empty(&fc_txn->changes));
	Assert(fc_txn->nentries == 0);
	Assert(fc_txn->nentries_mem == 0);
}

/*
 * 内存中变化的大小。
 */
static Size fc_ReorderBufferChangeSize(ReorderBufferChange *fc_change)
{
	Size		fc_sz = sizeof(ReorderBufferChange);

	switch (fc_change->action)
	{
			/* 跳过这些，它们都足够相似 */
		case REORDER_BUFFER_CHANGE_INSERT:
		case REORDER_BUFFER_CHANGE_UPDATE:
		case REORDER_BUFFER_CHANGE_DELETE:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
			{
				ReorderBufferTupleBuf *fc_oldtup,
						   *fc_newtup;
				Size		fc_oldlen = 0;
				Size		fc_newlen = 0;

				fc_oldtup = fc_change->data.tp.oldtuple;
				fc_newtup = fc_change->data.tp.newtuple;

				if (fc_oldtup)
				{
					fc_sz += sizeof(HeapTupleData);
					fc_oldlen = fc_oldtup->tuple.t_len;
					fc_sz += fc_oldlen;
				}

				if (fc_newtup)
				{
					fc_sz += sizeof(HeapTupleData);
					fc_newlen = fc_newtup->tuple.t_len;
					fc_sz += fc_newlen;
				}

				break;
			}
		case REORDER_BUFFER_CHANGE_MESSAGE:
			{
				Size		fc_prefix_size = strlen(fc_change->data.msg.prefix) + 1;

				fc_sz += fc_prefix_size + fc_change->data.msg.message_size +
					sizeof(Size) + sizeof(Size);

				break;
			}
		case REORDER_BUFFER_CHANGE_INVALIDATION:
			{
				fc_sz += sizeof(SharedInvalidationMessage) *
					fc_change->data.inval.ninvalidations;
				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
			{
				Snapshot	fc_snap;

				fc_snap = fc_change->data.snapshot;

				fc_sz += sizeof(SnapshotData) +
					sizeof(TransactionId) * fc_snap->xcnt +
					sizeof(TransactionId) * fc_snap->subxcnt;

				break;
			}
		case REORDER_BUFFER_CHANGE_TRUNCATE:
			{
				fc_sz += sizeof(Oid) * fc_change->data.truncate.nrelids;

				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
			/* ReorderBufferChange 包含所有重要内容 */
			break;
	}

	return fc_sz;
}


/*
 * 将若干溢出到磁盘的更改恢复到内存中。
 */
static Size fc_ReorderBufferRestoreChanges(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
							TXNEntryFile *fc_file, XLogSegNo *fc_segno)
{
	Size		fc_restored = 0;
	XLogSegNo	fc_last_segno;
	dlist_mutable_iter fc_cleanup_iter;
	File	   *fc_fd = &fc_file->vfd;

	Assert(fc_txn->first_lsn != InvalidXLogRecPtr);
	Assert(fc_txn->final_lsn != InvalidXLogRecPtr);

	/* 释放当前条目，以便为更多条目腾出内存 */
	dlist_foreach_modify(fc_cleanup_iter, &fc_txn->changes)
	{
		ReorderBufferChange *fc_cleanup =
		dlist_container(ReorderBufferChange, node, fc_cleanup_iter.cur);

		dlist_delete(&fc_cleanup->node);
		ReorderBufferReturnChange(fc_rb, fc_cleanup, true);
	}
	fc_txn->nentries_mem = 0;
	Assert(dlist_is_empty(&fc_txn->changes));

	XLByteToSeg(fc_txn->final_lsn, fc_last_segno, wal_segment_size);

	while (fc_restored < max_changes_in_memory && *fc_segno <= fc_last_segno)
	{
		int			fc_readBytes;
		ReorderBufferDiskChange *fc_ondisk;

		CHECK_FOR_INTERRUPTS();

		if (*fc_fd == -1)
		{
			char		fc_path[MAXPGPATH];

			/* 第一次进入 */
			if (*fc_segno == 0)
				XLByteToSeg(fc_txn->first_lsn, *fc_segno, wal_segment_size);

			Assert(*fc_segno != 0 || dlist_is_empty(&fc_txn->changes));

			/*
			 * 这里不需要关心 TLI，只在单次运行中使用，因此每个 LSN
			 * 只映射到特定的 WAL 记录。
			 */
			fc_ReorderBufferSerializedPath(fc_path, MyReplicationSlot, fc_txn->xid,
										*fc_segno);

			*fc_fd = PathNameOpenFile(fc_path, O_RDONLY | PG_BINARY);

			/* 即使在失败的情况下重置偏移量也没有坏处 */
			fc_file->curOffset = 0;

			if (*fc_fd < 0 && errno == ENOENT)
			{
				*fc_fd = -1;
				(*fc_segno)++;
				continue;
			}
			else if (*fc_fd < 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not open file \"%s\": %m",
								fc_path)));
		}

		/*
		 * 读取具有总大小信息的变化的静态大小部分。
		 * 如果我们无法读取记录，则我们已到达该文件的末尾。
		 */
		fc_ReorderBufferSerializeReserve(fc_rb, sizeof(ReorderBufferDiskChange));
		fc_readBytes = FileRead(fc_file->vfd, fc_rb->outbuf,
							 sizeof(ReorderBufferDiskChange),
							 fc_file->curOffset, WAIT_EVENT_REORDER_BUFFER_READ);

		/* eof */
		if (fc_readBytes == 0)
		{
			FileClose(*fc_fd);
			*fc_fd = -1;
			(*fc_segno)++;
			continue;
		}
		else if (fc_readBytes < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from reorderbuffer spill file: %m")));
		else if (fc_readBytes != sizeof(ReorderBufferDiskChange))
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
							fc_readBytes,
							(uint32) sizeof(ReorderBufferDiskChange))));

		fc_file->curOffset += fc_readBytes;

		fc_ondisk = (ReorderBufferDiskChange *) fc_rb->outbuf;

		fc_ReorderBufferSerializeReserve(fc_rb,
									  sizeof(ReorderBufferDiskChange) + fc_ondisk->size);
		fc_ondisk = (ReorderBufferDiskChange *) fc_rb->outbuf;

		fc_readBytes = FileRead(fc_file->vfd,
							 fc_rb->outbuf + sizeof(ReorderBufferDiskChange),
							 fc_ondisk->size - sizeof(ReorderBufferDiskChange),
							 fc_file->curOffset,
							 WAIT_EVENT_REORDER_BUFFER_READ);

		if (fc_readBytes < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from reorderbuffer spill file: %m")));
		else if (fc_readBytes != fc_ondisk->size - sizeof(ReorderBufferDiskChange))
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from reorderbuffer spill file: read %d instead of %u bytes",
							fc_readBytes,
							(uint32) (fc_ondisk->size - sizeof(ReorderBufferDiskChange)))));

		fc_file->curOffset += fc_readBytes;

		/*
		 * 好吧，从磁盘读取完整的更改，现在将其恢复为正确的
		 * 内存格式
		 */
		fc_ReorderBufferRestoreChange(fc_rb, fc_txn, fc_rb->outbuf);
		fc_restored++;
	}

	return fc_restored;
}

/*
 * 将更改从其磁盘格式转换为内存格式，并将其排入
 * TXN 的 ->changes 列表中。
 *
 * 注意：虽然“data”被声明为 char*，但在入口时它指向一个
 * maxalign 的缓冲区，这使得在该函数的大部分情况下，假设
 * 指向的数据对于直接访问是适当对齐的。
 */
static void fc_ReorderBufferRestoreChange(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
						   char *fc_data)
{
	ReorderBufferDiskChange *fc_ondisk;
	ReorderBufferChange *fc_change;

	fc_ondisk = (ReorderBufferDiskChange *) fc_data;

	fc_change = ReorderBufferGetChange(fc_rb);

	/* 复制静态部分 */
	memcpy(fc_change, &fc_ondisk->change, sizeof(ReorderBufferChange));

	fc_data += sizeof(ReorderBufferDiskChange);

	/* 恢复单独的内容 */
	switch (fc_change->action)
	{
			/* 跳过这些，它们都足够相似 */
		case REORDER_BUFFER_CHANGE_INSERT:
		case REORDER_BUFFER_CHANGE_UPDATE:
		case REORDER_BUFFER_CHANGE_DELETE:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT:
			if (fc_change->data.tp.oldtuple)
			{
				uint32		fc_tuplelen = ((HeapTuple) fc_data)->t_len;

				fc_change->data.tp.oldtuple =
					ReorderBufferGetTupleBuf(fc_rb, fc_tuplelen - SizeofHeapTupleHeader);

				/* 恢复 ->tuple */
				memcpy(&fc_change->data.tp.oldtuple->tuple, fc_data,
					   sizeof(HeapTupleData));
				fc_data += sizeof(HeapTupleData);

				/* 将 t_data 指针重置为新的 tuplebuf */
				fc_change->data.tp.oldtuple->tuple.t_data =
					ReorderBufferTupleBufData(fc_change->data.tp.oldtuple);

				/* 恢复元组数据本身 */
				memcpy(fc_change->data.tp.oldtuple->tuple.t_data, fc_data, fc_tuplelen);
				fc_data += fc_tuplelen;
			}

			if (fc_change->data.tp.newtuple)
			{
				/* 在这里，数据可能没有适当对齐！ */
				uint32		fc_tuplelen;

				memcpy(&fc_tuplelen, fc_data + offsetof(HeapTupleData, t_len),
					   sizeof(uint32));

				fc_change->data.tp.newtuple =
					ReorderBufferGetTupleBuf(fc_rb, fc_tuplelen - SizeofHeapTupleHeader);

				/* 恢复 ->tuple */
				memcpy(&fc_change->data.tp.newtuple->tuple, fc_data,
					   sizeof(HeapTupleData));
				fc_data += sizeof(HeapTupleData);

				/* 将 t_data 指针重置为新的 tuplebuf */
				fc_change->data.tp.newtuple->tuple.t_data =
					ReorderBufferTupleBufData(fc_change->data.tp.newtuple);

				/* 恢复元组数据本身 */
				memcpy(fc_change->data.tp.newtuple->tuple.t_data, fc_data, fc_tuplelen);
				fc_data += fc_tuplelen;
			}

			break;
		case REORDER_BUFFER_CHANGE_MESSAGE:
			{
				Size		fc_prefix_size;

				/* 读取前缀 */
				memcpy(&fc_prefix_size, fc_data, sizeof(Size));
				fc_data += sizeof(Size);
				fc_change->data.msg.prefix = MemoryContextAlloc(fc_rb->context,
															 fc_prefix_size);
				memcpy(fc_change->data.msg.prefix, fc_data, fc_prefix_size);
				Assert(fc_change->data.msg.prefix[fc_prefix_size - 1] == '\0');
				fc_data += fc_prefix_size;

				/* 读取消息 */
				memcpy(&fc_change->data.msg.message_size, fc_data, sizeof(Size));
				fc_data += sizeof(Size);
				fc_change->data.msg.message = MemoryContextAlloc(fc_rb->context,
															  fc_change->data.msg.message_size);
				memcpy(fc_change->data.msg.message, fc_data,
					   fc_change->data.msg.message_size);
				fc_data += fc_change->data.msg.message_size;

				break;
			}
		case REORDER_BUFFER_CHANGE_INVALIDATION:
			{
				Size		fc_inval_size = sizeof(SharedInvalidationMessage) *
				fc_change->data.inval.ninvalidations;

				fc_change->data.inval.invalidations =
					MemoryContextAlloc(fc_rb->context, fc_inval_size);

				/* 读取消息 */
				memcpy(fc_change->data.inval.invalidations, fc_data, fc_inval_size);

				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
			{
				Snapshot	fc_oldsnap;
				Snapshot	fc_newsnap;
				Size		fc_size;

				fc_oldsnap = (Snapshot) fc_data;

				fc_size = sizeof(SnapshotData) +
					sizeof(TransactionId) * fc_oldsnap->xcnt +
					sizeof(TransactionId) * (fc_oldsnap->subxcnt + 0);

				fc_change->data.snapshot = MemoryContextAllocZero(fc_rb->context, fc_size);

				fc_newsnap = fc_change->data.snapshot;

				memcpy(fc_newsnap, fc_data, fc_size);
				fc_newsnap->xip = (TransactionId *)
					(((char *) fc_newsnap) + sizeof(SnapshotData));
				fc_newsnap->subxip = fc_newsnap->xip + fc_newsnap->xcnt;
				fc_newsnap->copied = true;
				break;
			}
			/* 基本结构包含所有数据，简单明了 */
		case REORDER_BUFFER_CHANGE_TRUNCATE:
			{
				Oid		   *fc_relids;

				fc_relids = ReorderBufferGetRelids(fc_rb,
												fc_change->data.truncate.nrelids);
				memcpy(fc_relids, fc_data, fc_change->data.truncate.nrelids * sizeof(Oid));
				fc_change->data.truncate.relids = fc_relids;

				break;
			}
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
			break;
	}

	dlist_push_tail(&fc_txn->changes, &fc_change->node);
	fc_txn->nentries_mem++;

	/*
	 * 更新恢复更改的内存账务。虽然在恢复更改时我们并不检查
	 * 内存限制（我们只在解码后初始排队更改时这样做），但是我们
	 * 仍然需要这样做，因为我们稍后会释放这些更改，这也会更新
	 * 账务（从计数器中减去大小）。我们不希望在这里出现下溢。
	 */
	fc_ReorderBufferChangeMemoryUpdate(fc_rb, fc_change, true,
									fc_ReorderBufferChangeSize(fc_change));
}

/*
 * 删除传入事务的所有磁盘存储。
 */
static void fc_ReorderBufferRestoreCleanup(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	XLogSegNo	fc_first;
	XLogSegNo	fc_cur;
	XLogSegNo	fc_last;

	Assert(fc_txn->first_lsn != InvalidXLogRecPtr);
	Assert(fc_txn->final_lsn != InvalidXLogRecPtr);

	XLByteToSeg(fc_txn->first_lsn, fc_first, wal_segment_size);
	XLByteToSeg(fc_txn->final_lsn, fc_last, wal_segment_size);

	/* 迭代所有可能的文件名，并删除它们 */
	for (fc_cur = fc_first; fc_cur <= fc_last; fc_cur++)
	{
		char		fc_path[MAXPGPATH];

		fc_ReorderBufferSerializedPath(fc_path, MyReplicationSlot, fc_txn->xid, fc_cur);
		if (unlink(fc_path) != 0 && errno != ENOENT)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not remove file \"%s\": %m", fc_path)));
	}
}

/*
 * 在先前崩溃或解码会话退出后，从插槽目录中删除任何剩余的序列化重排缓冲区。
 */
static void fc_ReorderBufferCleanupSerializedTXNs(const char *fc_slotname)
{
	DIR		   *fc_spill_dir;
	struct dirent *fc_spill_de;
	struct stat fc_statbuf;
	char		fc_path[MAXPGPATH * 2 + 12];

	sprintf(fc_path, "pg_replslot/%s", fc_slotname);

	/* 我们这里只处理目录，如果不是我们的，则跳过 */
	if (lstat(fc_path, &fc_statbuf) == 0 && !S_ISDIR(fc_statbuf.st_mode))
		return;

	fc_spill_dir = AllocateDir(fc_path);
	while ((fc_spill_de = ReadDirExtended(fc_spill_dir, fc_path, INFO)) != NULL)
	{
		/* 仅查看可能是我们的名称 */
		if (strncmp(fc_spill_de->d_name, "xid", 3) == 0)
		{
			snprintf(fc_path, sizeof(fc_path),
					 "pg_replslot/%s/%s", fc_slotname,
					 fc_spill_de->d_name);

			if (unlink(fc_path) != 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not remove file \"%s\" during removal of pg_replslot/%s/xid*: %m",
								fc_path, fc_slotname)));
		}
	}
	FreeDir(fc_spill_dir);
}

/*
 * 给定复制槽、事务 ID 和段号，将相应的溢出文件填充到 'path' 中，
 * 这是一个调用者拥有的至少为 MAXPGPATH 大小的缓冲区。
 */
static void fc_ReorderBufferSerializedPath(char *fc_path, ReplicationSlot *fc_slot, TransactionId fc_xid,
							XLogSegNo fc_segno)
{
	XLogRecPtr	fc_recptr;

	XLogSegNoOffsetToRecPtr(fc_segno, 0, wal_segment_size, fc_recptr);

	snprintf(fc_path, MAXPGPATH, "pg_replslot/%s/xid-%u-lsn-%X-%X.spill",
			 NameStr(MyReplicationSlot->data.name),
			 fc_xid, LSN_FORMAT_ARGS(fc_recptr));
}

/*
 * 删除在我们重新启动/崩溃后溢出到磁盘的所有数据。
 * 在重新使用相关槽时将重新创建它。
 */
void StartupReorderBuffer(void)
{
	DIR		   *fc_logical_dir;
	struct dirent *fc_logical_de;

	fc_logical_dir = AllocateDir("pg_replslot");
	while ((fc_logical_de = ReadDir(fc_logical_dir, "pg_replslot")) != NULL)
	{
		if (strcmp(fc_logical_de->d_name, ".") == 0 ||
			strcmp(fc_logical_de->d_name, "..") == 0)
			continue;

		/* 如果它不能是槽，则跳过该目录 */
		if (!ReplicationSlotValidateName(fc_logical_de->d_name, DEBUG2))
			continue;

		/*
		 * 好的，必须是一个存活的逻辑槽，迭代并删除
		 * 所有以 xid-* 开头的内容
		 */
		fc_ReorderBufferCleanupSerializedTXNs(fc_logical_de->d_name);
	}
	FreeDir(fc_logical_dir);
}

/* ---------------------------------------
 * toast 重新组装支持
 * ---------------------------------------
 */

/*
 * 初始化每个元组的 toast 重构支持。
 */
static void fc_ReorderBufferToastInitHash(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	HASHCTL		fc_hash_ctl;

	Assert(fc_txn->toast_hash == NULL);

	fc_hash_ctl.keysize = sizeof(Oid);
	fc_hash_ctl.entrysize = sizeof(ReorderBufferToastEnt);
	fc_hash_ctl.hcxt = fc_rb->context;
	fc_txn->toast_hash = hash_create("ReorderBufferToastHash", 5, &fc_hash_ctl,
								  HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
}

/*
 * toast 重构的每个 toast-chunk 处理
 *
 * 附加一个 toast chunk，以便在元组 "拥有" 的
 * 已烤制 Datum 到来时，我们可以重构它。
 */
static void fc_ReorderBufferToastAppendChunk(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
							  Relation fc_relation, ReorderBufferChange *fc_change)
{
	ReorderBufferToastEnt *fc_ent;
	ReorderBufferTupleBuf *fc_newtup;
	bool		fc_found;
	int32		fc_chunksize;
	bool		fc_isnull;
	Pointer		fc_chunk;
	TupleDesc	fc_desc = RelationGetDescr(fc_relation);
	Oid			fc_chunk_id;
	int32		fc_chunk_seq;

	if (fc_txn->toast_hash == NULL)
		fc_ReorderBufferToastInitHash(fc_rb, fc_txn);

	Assert(IsToastRelation(fc_relation));

	fc_newtup = fc_change->data.tp.newtuple;
	fc_chunk_id = DatumGetObjectId(fastgetattr(&fc_newtup->tuple, 1, fc_desc, &fc_isnull));
	Assert(!fc_isnull);
	fc_chunk_seq = DatumGetInt32(fastgetattr(&fc_newtup->tuple, 2, fc_desc, &fc_isnull));
	Assert(!fc_isnull);

	fc_ent = (ReorderBufferToastEnt *)
		hash_search(fc_txn->toast_hash,
					(void *) &fc_chunk_id,
					HASH_ENTER,
					&fc_found);

	if (!fc_found)
	{
		Assert(fc_ent->chunk_id == fc_chunk_id);
		fc_ent->num_chunks = 0;
		fc_ent->last_chunk_seq = 0;
		fc_ent->size = 0;
		fc_ent->reconstructed = NULL;
		dlist_init(&fc_ent->chunks);

		if (fc_chunk_seq != 0)
			elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq 0",
				 fc_chunk_seq, fc_chunk_id);
	}
	else if (fc_found && fc_chunk_seq != fc_ent->last_chunk_seq + 1)
		elog(ERROR, "got sequence entry %d for toast chunk %u instead of seq %d",
			 fc_chunk_seq, fc_chunk_id, fc_ent->last_chunk_seq + 1);

	fc_chunk = DatumGetPointer(fastgetattr(&fc_newtup->tuple, 3, fc_desc, &fc_isnull));
	Assert(!fc_isnull);

	/* 计算大小，以便我们之后可以一次性分配正确的大小 */
	if (!VARATT_IS_EXTENDED(fc_chunk))
		fc_chunksize = VARSIZE(fc_chunk) - VARHDRSZ;
	else if (VARATT_IS_SHORT(fc_chunk))
		/* 由于 heap_form_tuple 做它的工作而可能发生 */
		fc_chunksize = VARSIZE_SHORT(fc_chunk) - VARHDRSZ_SHORT;
	else
		elog(ERROR, "unexpected type of toast chunk");

	fc_ent->size += fc_chunksize;
	fc_ent->last_chunk_seq = fc_chunk_seq;
	fc_ent->num_chunks++;
	dlist_push_tail(&fc_ent->chunks, &fc_change->node);
}

/*
 * 重新调整 change->newtuple 指向内存中的 toast 元组，而不是
 * 可能不再存在的磁盘上的 toast 元组（想想 DROP TABLE 或 VACUUM）。
 *
 * 然而，我们不能替换未更改的 toast 元组，因此那些仍然会指向
 * 磁盘上的 toast 数据。
 *
 * 在用去烤元组数据更新现有更改时，我们需要更新内存计数信息，
 * 因为更改大小会有所不同。
 * 否则，会导致计数不同步，从而在意外时间触发序列化。
 *
 * 我们简单地在重新调整元组之前减去更改的大小，
 * 然后再添加新大小。这使得看起来更改被移除
 * 然后又被添加回来，但它只是调整计数信息。
 *
 * 特别是它不会触发序列化，反正这是毫无意义的，
 * 因为这发生在提交处理期间，就在将更改
 * 交给输出插件之前。
 */
static void fc_ReorderBufferToastReplace(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn,
						  Relation fc_relation, ReorderBufferChange *fc_change)
{
	TupleDesc	fc_desc;
	int			fc_natt;
	Datum	   *fc_attrs;
	bool	   *fc_isnull;
	bool	   *fc_free;
	HeapTuple	fc_tmphtup;
	Relation	fc_toast_rel;
	TupleDesc	fc_toast_desc;
	MemoryContext fc_oldcontext;
	ReorderBufferTupleBuf *fc_newtup;
	Size		fc_old_size;

	/* 未更改的 toast 元组 */
	if (fc_txn->toast_hash == NULL)
		return;

	/*
	 * 我们将修改更改的大小。因此，为了确保
	 * 记账正确，我们记录当前的更改大小，
	 * 然后在重新计算更改后，我们将减去记录的大小，
	 * 然后在最后重新添加新的更改大小。
	 * 我们不会立即减去旧大小，因为如果在添加新大小之前
	 * 发生任何错误，我们将释放更改，这将更新计数信息
	 * （从计数器中减去大小）。我们不希望在此处出现下溢。
	 */
	fc_old_size = fc_ReorderBufferChangeSize(fc_change);

	fc_oldcontext = MemoryContextSwitchTo(fc_rb->context);

	/* 我们应该只在 INSERT 或 UPDATE 中有 toast 元组 */
	Assert(fc_change->data.tp.newtuple);

	fc_desc = RelationGetDescr(fc_relation);

	fc_toast_rel = RelationIdGetRelation(fc_relation->rd_rel->reltoastrelid);
	if (!RelationIsValid(fc_toast_rel))
		elog(ERROR, "could not open toast relation with OID %u (base relation \"%s\")",
			 fc_relation->rd_rel->reltoastrelid, RelationGetRelationName(fc_relation));

	fc_toast_desc = RelationGetDescr(fc_toast_rel);

	/* 我们应该从栈中分配内存吗？ */
	fc_attrs = palloc0(sizeof(Datum) * fc_desc->natts);
	fc_isnull = palloc0(sizeof(bool) * fc_desc->natts);
	fc_free = palloc0(sizeof(bool) * fc_desc->natts);

	fc_newtup = fc_change->data.tp.newtuple;

	heap_deform_tuple(&fc_newtup->tuple, fc_desc, fc_attrs, fc_isnull);

	for (fc_natt = 0; fc_natt < fc_desc->natts; fc_natt++)
	{
		Form_pg_attribute fc_attr = TupleDescAttr(fc_desc, fc_natt);
		ReorderBufferToastEnt *fc_ent;
		struct varlena *fc_varlena;

		/* va_rawsize是原始数据的大小--包括头部 */
		struct varatt_external fc_toast_pointer;
		struct varatt_indirect fc_redirect_pointer;
		struct varlena *fc_new_datum = NULL;
		struct varlena *fc_reconstructed;
		dlist_iter	fc_it;
		Size		fc_data_done = 0;

		/* 系统列未被烤制 */
		if (fc_attr->attnum < 0)
			continue;

		if (fc_attr->attisdropped)
			continue;

		/* 不是 varlena 数据类型 */
		if (fc_attr->attlen != -1)
			continue;

		/* 无数据 */
		if (fc_isnull[fc_natt])
			continue;

		/* 好的，我们知道我们有一个 toast datum */
		fc_varlena = (struct varlena *) DatumGetPointer(fc_attrs[fc_natt]);

		/* 如果元组不是外部的，就不需要做任何事情 */
		if (!VARATT_IS_EXTERNAL(fc_varlena))
			continue;

		VARATT_EXTERNAL_GET_POINTER(fc_toast_pointer, fc_varlena);

		/*
		 * 检查 toast 元组是否更改，如果是，则替换。
		 */
		fc_ent = (ReorderBufferToastEnt *)
			hash_search(fc_txn->toast_hash,
						(void *) &fc_toast_pointer.va_valueid,
						HASH_FIND,
						NULL);
		if (fc_ent == NULL)
			continue;

		fc_new_datum =
			(struct varlena *) palloc0(INDIRECT_POINTER_SIZE);

		fc_free[fc_natt] = true;

		fc_reconstructed = palloc0(fc_toast_pointer.va_rawsize);

		fc_ent->reconstructed = fc_reconstructed;

		/* 将 toast 元组从其部分重新拼接在一起 */
		dlist_foreach(fc_it, &fc_ent->chunks)
		{
			bool		fc_isnull;
			ReorderBufferChange *fc_cchange;
			ReorderBufferTupleBuf *fc_ctup;
			Pointer		fc_chunk;

			fc_cchange = dlist_container(ReorderBufferChange, node, fc_it.cur);
			fc_ctup = fc_cchange->data.tp.newtuple;
			fc_chunk = DatumGetPointer(fastgetattr(&fc_ctup->tuple, 3, fc_toast_desc, &fc_isnull));

			Assert(!fc_isnull);
			Assert(!VARATT_IS_EXTERNAL(fc_chunk));
			Assert(!VARATT_IS_SHORT(fc_chunk));

			memcpy(VARDATA(fc_reconstructed) + fc_data_done,
				   VARDATA(fc_chunk),
				   VARSIZE(fc_chunk) - VARHDRSZ);
			fc_data_done += VARSIZE(fc_chunk) - VARHDRSZ;
		}
		Assert(fc_data_done == VARATT_EXTERNAL_GET_EXTSIZE(fc_toast_pointer));

		/* 确保标记为压缩或未压缩 */
		if (VARATT_EXTERNAL_IS_COMPRESSED(fc_toast_pointer))
			SET_VARSIZE_COMPRESSED(fc_reconstructed, fc_data_done + VARHDRSZ);
		else
			SET_VARSIZE(fc_reconstructed, fc_data_done + VARHDRSZ);

		memset(&fc_redirect_pointer, 0, sizeof(fc_redirect_pointer));
		fc_redirect_pointer.pointer = fc_reconstructed;

		SET_VARTAG_EXTERNAL(fc_new_datum, VARTAG_INDIRECT);
		memcpy(VARDATA_EXTERNAL(fc_new_datum), &fc_redirect_pointer,
			   sizeof(fc_redirect_pointer));

		fc_attrs[fc_natt] = PointerGetDatum(fc_new_datum);
	}

	/*
	 * 在单独的内存中构建元组，并将元组复制回传递给
	 * 输出插件的 tuplebuf 中。我们不能直接 heap_fill_tuple()
	 * 到 tuplebuf，因为 attrs[] 将指向当前内容。
	 */
	fc_tmphtup = heap_form_tuple(fc_desc, fc_attrs, fc_isnull);
	Assert(fc_newtup->tuple.t_len <= MaxHeapTupleSize);
	Assert(ReorderBufferTupleBufData(fc_newtup) == fc_newtup->tuple.t_data);

	memcpy(fc_newtup->tuple.t_data, fc_tmphtup->t_data, fc_tmphtup->t_len);
	fc_newtup->tuple.t_len = fc_tmphtup->t_len;

	/*
	 * 释放我们不会再需要的资源，更持久的内容将在
	 * ReorderBufferToastReset() 中释放。
	 */
	RelationClose(fc_toast_rel);
	pfree(fc_tmphtup);
	for (fc_natt = 0; fc_natt < fc_desc->natts; fc_natt++)
	{
		if (fc_free[fc_natt])
			pfree(DatumGetPointer(fc_attrs[fc_natt]));
	}
	pfree(fc_attrs);
	pfree(fc_free);
	pfree(fc_isnull);

	MemoryContextSwitchTo(fc_oldcontext);

	/* 减去旧的变更大小 */
	fc_ReorderBufferChangeMemoryUpdate(fc_rb, fc_change, false, fc_old_size);
	/* 现在按照正确的大小重新添加变更 */
	fc_ReorderBufferChangeMemoryUpdate(fc_rb, fc_change, true,
									fc_ReorderBufferChangeSize(fc_change));
}

/*
 * 释放所有为吐司重构分配的资源。
 */
static void fc_ReorderBufferToastReset(ReorderBuffer *fc_rb, ReorderBufferTXN *fc_txn)
{
	HASH_SEQ_STATUS fc_hstat;
	ReorderBufferToastEnt *fc_ent;

	if (fc_txn->toast_hash == NULL)
		return;

	/* 按顺序遍历哈希并释放所有内容 */
	hash_seq_init(&fc_hstat, fc_txn->toast_hash);
	while ((fc_ent = (ReorderBufferToastEnt *) hash_seq_search(&fc_hstat)) != NULL)
	{
		dlist_mutable_iter fc_it;

		if (fc_ent->reconstructed != NULL)
			pfree(fc_ent->reconstructed);

		dlist_foreach_modify(fc_it, &fc_ent->chunks)
		{
			ReorderBufferChange *fc_change =
			dlist_container(ReorderBufferChange, node, fc_it.cur);

			dlist_delete(&fc_change->node);
			ReorderBufferReturnChange(fc_rb, fc_change, true);
		}
	}

	hash_destroy(fc_txn->toast_hash);
	fc_txn->toast_hash = NULL;
}


/* ---------------------------------------
 * 逻辑解码的可见性支持
 *
 *
 * 使用解码快照时查找实际的 cmin/cmax 值。我们不能
 * 总是依赖存储的 cmin/cmax 值，因为有两种情况：
 *
 * * 一个元组在一个事务中被多次更改，因此
 *   具有组合 CID。组合 CID 仅在
 *   单个事务的持续时间内有效。
 * * 一个具有 cmin 但没有 cmax (因此没有组合 CID) 的元组在
 *   创建它的另一个事务中被删除/更新，这个事务是我们
 *   现在正在查看的。由于 cmin、cmax 或组合 CID 中只有一个
 *   实际存储在堆中，我们不再可以访问我们所需的值。
 *
 * 为了解决这些问题，我们为 (relfilenode, ctid) 键的
 * (cmin, cmax) 元组维护每个事务的哈希表，该哈希表包含实际的
 * (cmin, cmax) 值。这也通过根本不关心组合 CID 来处理组合 CID。
 * 由于我们有真实的 cmin/cmax 值，组合 CID 就不再重要。
 *
 * 由于我们只关心目录元组，这个
 * 哈希表的开销应该是可以接受的。
 *
 * 堆重写使这一点变得复杂了一些，查看 rewriteheap.c 获取
 * 详细信息。
 * -------------------------------------------------------------------------
 */

/* 用于按 LSN 高效排序映射文件的结构 */
typedef struct RewriteMappingFile
{
	XLogRecPtr	lsn;
	char		fname[MAXPGPATH];
} RewriteMappingFile;

#ifdef NOT_USED
static void fc_DisplayMapping(HTAB *fc_tuplecid_data)
{
	HASH_SEQ_STATUS fc_hstat;
	ReorderBufferTupleCidEnt *fc_ent;

	hash_seq_init(&fc_hstat, fc_tuplecid_data);
	while ((fc_ent = (ReorderBufferTupleCidEnt *) hash_seq_search(&fc_hstat)) != NULL)
	{
		elog(DEBUG3, "mapping: node: %u/%u/%u tid: %u/%u cmin: %u, cmax: %u",
			 fc_ent->key.relnode.dbNode,
			 fc_ent->key.relnode.spcNode,
			 fc_ent->key.relnode.relNode,
			 ItemPointerGetBlockNumber(&fc_ent->key.tid),
			 ItemPointerGetOffsetNumber(&fc_ent->key.tid),
			 fc_ent->cmin,
			 fc_ent->cmax
			);
	}
}
#endif

/*
 * 将单个映射文件应用于 tuplecid_data。
 *
 * 映射文件必须经过验证 a) 已提交 b) 针对我们的
 * 事务 c) 按 LSN 顺序应用。
 */
static void fc_ApplyLogicalMappingFile(HTAB *fc_tuplecid_data, Oid fc_relid, const char *fc_fname)
{
	char		fc_path[MAXPGPATH];
	int			fc_fd;
	int			fc_readBytes;
	LogicalRewriteMappingData fc_map;

	sprintf(fc_path, "pg_logical/mappings/%s", fc_fname);
	fc_fd = OpenTransientFile(fc_path, O_RDONLY | PG_BINARY);
	if (fc_fd < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not open file \"%s\": %m", fc_path)));

	while (true)
	{
		ReorderBufferTupleCidKey fc_key;
		ReorderBufferTupleCidEnt *fc_ent;
		ReorderBufferTupleCidEnt *fc_new_ent;
		bool		fc_found;

		/* 注意填充问题 */
		memset(&fc_key, 0, sizeof(ReorderBufferTupleCidKey));

		/* 读取所有映射直到文件末尾 */
		pgstat_report_wait_start(WAIT_EVENT_REORDER_LOGICAL_MAPPING_READ);
		fc_readBytes = read(fc_fd, &fc_map, sizeof(LogicalRewriteMappingData));
		pgstat_report_wait_end();

		if (fc_readBytes < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read file \"%s\": %m",
							fc_path)));
		else if (fc_readBytes == 0)	/* EOF */
			break;
		else if (fc_readBytes != sizeof(LogicalRewriteMappingData))
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not read from file \"%s\": read %d instead of %d bytes",
							fc_path, fc_readBytes,
							(int32) sizeof(LogicalRewriteMappingData))));

		fc_key.relnode = fc_map.old_node;
		ItemPointerCopy(&fc_map.old_tid,
						&fc_key.tid);


		fc_ent = (ReorderBufferTupleCidEnt *)
			hash_search(fc_tuplecid_data,
						(void *) &fc_key,
						HASH_FIND,
						NULL);

		/* 无现有映射，无需更新 */
		if (!fc_ent)
			continue;

		fc_key.relnode = fc_map.new_node;
		ItemPointerCopy(&fc_map.new_tid,
						&fc_key.tid);

		fc_new_ent = (ReorderBufferTupleCidEnt *)
			hash_search(fc_tuplecid_data,
						(void *) &fc_key,
						HASH_ENTER,
						&fc_found);

		if (fc_found)
		{
			/*
			 * 确保现有映射是合理的。我们有时在重写期间更新
			 * 还没有 cmax 的旧记录（例如 pg_class 自身
			 * 在重写时的条目），因此允许这种情况。
			 */
			Assert(fc_ent->cmin == InvalidCommandId || fc_ent->cmin == fc_new_ent->cmin);
			Assert(fc_ent->cmax == InvalidCommandId || fc_ent->cmax == fc_new_ent->cmax);
		}
		else
		{
			/* 更新映射 */
			fc_new_ent->cmin = fc_ent->cmin;
			fc_new_ent->cmax = fc_ent->cmax;
			fc_new_ent->combocid = fc_ent->combocid;
		}
	}

	if (CloseTransientFile(fc_fd) != 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not close file \"%s\": %m", fc_path)));
}


/*
 * 检查交易 ID 'xid' 是否在预排序数组 'xip' 中。
 */
static bool fc_TransactionIdInArray(TransactionId fc_xid, TransactionId *fc_xip, Size fc_num)
{
	return bsearch(&fc_xid, fc_xip, fc_num,
				   sizeof(TransactionId), xidComparator) != NULL;
}

/*
 * list_sort() 比较器用于按 LSN 顺序排序 RewriteMappingFiles。
 */
static int fc_file_sort_by_lsn(const ListCell *fc_a_p, const ListCell *fc_b_p)
{
	RewriteMappingFile *fc_a = (RewriteMappingFile *) lfirst(fc_a_p);
	RewriteMappingFile *fc_b = (RewriteMappingFile *) lfirst(fc_b_p);

	if (fc_a->lsn < fc_b->lsn)
		return -1;
	else if (fc_a->lsn > fc_b->lsn)
		return 1;
	return 0;
}

/*
 * 如果有任何针对我们的事务的逻辑重映射文件，请应用它们
 * 以 relid 为目标。
 */
static void fc_UpdateLogicalMappings(HTAB *fc_tuplecid_data, Oid fc_relid, Snapshot fc_snapshot)
{
	DIR		   *fc_mapping_dir;
	struct dirent *fc_mapping_de;
	List	   *fc_files = NIL;
	ListCell   *fc_file;
	Oid			fc_dboid = IsSharedRelation(fc_relid) ? InvalidOid : MyDatabaseId;

	fc_mapping_dir = AllocateDir("pg_logical/mappings");
	while ((fc_mapping_de = ReadDir(fc_mapping_dir, "pg_logical/mappings")) != NULL)
	{
		Oid			fc_f_dboid;
		Oid			fc_f_relid;
		TransactionId fc_f_mapped_xid;
		TransactionId fc_f_create_xid;
		XLogRecPtr	fc_f_lsn;
		uint32		fc_f_hi,
					fc_f_lo;
		RewriteMappingFile *fc_f;

		if (strcmp(fc_mapping_de->d_name, ".") == 0 ||
			strcmp(fc_mapping_de->d_name, "..") == 0)
			continue;

		/* 忽略不是我们的文件 */
		if (strncmp(fc_mapping_de->d_name, "map-", 4) != 0)
			continue;

		if (sscanf(fc_mapping_de->d_name, LOGICAL_REWRITE_FORMAT,
				   &fc_f_dboid, &fc_f_relid, &fc_f_hi, &fc_f_lo,
				   &fc_f_mapped_xid, &fc_f_create_xid) != 6)
			elog(ERROR, "could not parse filename \"%s\"", fc_mapping_de->d_name);

		fc_f_lsn = ((uint64) fc_f_hi) << 32 | fc_f_lo;

		/* 另一个数据库的映射 */
		if (fc_f_dboid != fc_dboid)
			continue;

		/* 另一个关系的映射 */
		if (fc_f_relid != fc_relid)
			continue;

		/* 创建交易是否中止？ */
		if (!TransactionIdDidCommit(fc_f_create_xid))
			continue;

		/* 不适用于我们的事务 */
		if (!fc_TransactionIdInArray(fc_f_mapped_xid, fc_snapshot->subxip, fc_snapshot->subxcnt))
			continue;

		/* 好的，相关，排队应用 */
		fc_f = palloc(sizeof(RewriteMappingFile));
		fc_f->lsn = fc_f_lsn;
		strcpy(fc_f->fname, fc_mapping_de->d_name);
		fc_files = lappend(fc_files, fc_f);
	}
	FreeDir(fc_mapping_dir);

	/* 对文件进行排序，以便我们按 LSN 顺序应用它们 */
	list_sort(fc_files, fc_file_sort_by_lsn);

	foreach(fc_file, fc_files)
	{
		RewriteMappingFile *fc_f = (RewriteMappingFile *) lfirst(fc_file);

		elog(DEBUG1, "applying mapping: \"%s\" in %u", fc_f->fname,
			 fc_snapshot->subxip[0]);
		fc_ApplyLogicalMappingFile(fc_tuplecid_data, fc_relid, fc_f->fname);
		pfree(fc_f);
	}
}

/*
 * 在逻辑解码期间查找元组的 cmin/cmax，此时我们不能依赖
 * 组合 CID。
 */
bool ResolveCminCmaxDuringDecoding(HTAB *fc_tuplecid_data,
							  Snapshot fc_snapshot,
							  HeapTuple fc_htup, Buffer fc_buffer,
							  CommandId *fc_cmin, CommandId *fc_cmax)
{
	ReorderBufferTupleCidKey fc_key;
	ReorderBufferTupleCidEnt *fc_ent;
	ForkNumber	fc_forkno;
	BlockNumber fc_blockno;
	bool		fc_updated_mapping = false;

	/*
	 * 如果tuplecid_data无效，则返回未解决状态。因为在流式传输进行中的事务时，我们可能会遇到在实际解码之前的CID的元组。例如，考虑INSERT后跟TRUNCATE的情况，在应用INSERT时，TRUNCATE可能尚未解码。因此，在这种情况下，我们假设CID来自未来的命令。
	 */
	if (fc_tuplecid_data == NULL)
		return false;

	/* 注意填充问题 */
	memset(&fc_key, 0, sizeof(fc_key));

	Assert(!BufferIsLocal(fc_buffer));

	/*
	 * 从缓冲区获取relfilenode，没有其他方便的方式访问它。
	 */
	BufferGetTag(fc_buffer, &fc_key.relnode, &fc_forkno, &fc_blockno);

	/* 元组只能在主分支中 */
	Assert(fc_forkno == MAIN_FORKNUM);
	Assert(fc_blockno == ItemPointerGetBlockNumber(&fc_htup->t_self));

	ItemPointerCopy(&fc_htup->t_self,
					&fc_key.tid);

restart:
	fc_ent = (ReorderBufferTupleCidEnt *)
		hash_search(fc_tuplecid_data,
					(void *) &fc_key,
					HASH_FIND,
					NULL);

	/*
	 * 未找到映射，检查表是否已重写，如果是，则应用映射，但只能执行一次 - 在此期间不能有新的映射，因为我们必须对关系保持锁定。
	 */
	if (fc_ent == NULL && !fc_updated_mapping)
	{
		fc_UpdateLogicalMappings(fc_tuplecid_data, fc_htup->t_tableOid, fc_snapshot);
		/* 现在检查但不要再次更新映射 */
		fc_updated_mapping = true;
		goto restart;
	}
	else if (fc_ent == NULL)
		return false;

	if (fc_cmin)
		*fc_cmin = fc_ent->cmin;
	if (fc_cmax)
		*fc_cmax = fc_ent->cmax;
	return true;
}
